?
模型结构
消息发送流程
1.生产者把消息发送给交换机
2.交换机把消息发送给每个绑定的临时队列
3.消费者通过队列消费消息(每一个消费者都绑定一个临时队列)
引入依赖
<!-- 引入rabbitmq依赖-->
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>5.7.2</version>
</dependency>
代码编写
1.mq获取连接的工具类
package utils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class RabbitMqUtils {
private static ConnectionFactory connectionFactory;
static {
connectionFactory = new ConnectionFactory();
connectionFactory.setHost("192.168.xxx.xxx");//服务器的ip
connectionFactory.setPort(5672);//访问端口号
connectionFactory.setVirtualHost("/lln");//虚拟主机名称
connectionFactory.setUsername("lln");//账号
connectionFactory.setPassword("123");//密码
}
//提供连接对象的方法
public static Connection getConnection(){
try{
return connectionFactory.newConnection();
}catch (Exception e){
e.printStackTrace();
}
return null;
}
//关闭通道和关闭连接工具的方法
public static void closeConnectionAndChanel(Channel channel,Connection connection){
try {
if (channel!=null){
channel.close();
}
if (connection!=null){
connection.close();
}
}catch (Exception e){
e.printStackTrace();
}
}
}
2.生产者
package fanout;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import utils.RabbitMqUtils;
import java.io.IOException;
public class Provider {
public static void main(String[] args) throws IOException {
//获取连接
Connection connection = RabbitMqUtils.getConnection();
Channel channel = connection.createChannel();
//将通道声明指定的交换机
//参数1:交换机名称,参数2:交换机类型 fanout为广播类型
channel.exchangeDeclare("logs","fanout");
//发送消息
channel.basicPublish("logs","",null,"fanout type message".getBytes());
RabbitMqUtils.closeConnectionAndChanel(channel,connection);
}
}
3.消费者(复制多个即可)
package fanout;
import com.rabbitmq.client.*;
import utils.RabbitMqUtils;
import java.io.IOException;
public class Consumer1 {
public static void main(String[] args) throws IOException {
Connection connection = RabbitMqUtils.getConnection();
Channel channel = connection.createChannel();
//通道绑定交换机
channel.exchangeDeclare("logs","fanout");
//临时队列
String queueName = channel.queueDeclare().getQueue();
//绑定交换机和队列
channel.queueBind(queueName,"logs","");
//消费消息
channel.basicConsume(queueName,true,new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("消费者1:"+new String(body));
}
});
}
}
运行效果
|