扇出模式 fanout 发布订阅模式
生产者
生产者发送消息到交换机（logs）,控制台输入消息作为生产者的消息发送
package com.esint.rabbitmq.work03;
import com.esint.rabbitmq.RabbitMQUtils;
import com.rabbitmq.client.Channel;
import java.util.Scanner;
public class EmitLog {
public static final String EXCHANGE_NAME = "logs";
public static void main(String[] args) throws Exception {
Channel channel = RabbitMQUtils.getChannel();
//声明交换机
channel.exchangeDeclare(EXCHANGE_NAME,"fanout");
Scanner scanner = new Scanner(System.in);
while(scanner.hasNext()){
String mes = scanner.next();
channel.basicPublish(EXCHANGE_NAME,"",null,mes.getBytes("UTF-8"));
System.out.println("生产者发出消息："+mes.toString());
}
}
}交换机如下：
发布者如下：
消费者
创建两个消费来同时消费生产者的发布消息。声明两个临时队列，与交换机做绑定，消费消息
消费者01
package com.esint.rabbitmq.work03;
import com.esint.rabbitmq.RabbitMQUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
/**
* 消费者01的消息接受
*/
public class ReceiveLog01 {
public static final String EXCHANGE_NAME = "logs";
public static void main(String[] args) throws Exception {
//获取信道
Channel channel = RabbitMQUtils.getChannel();
//声明一个交换机
channel.exchangeDeclare(EXCHANGE_NAME,"fanout");
/**
*声明一个队列 队列名字随机
* 消费者断开与队列的链接后 队列自动删除
*/
String queue = channel.queueDeclare().getQueue();
/**
* 绑定交换机与队列
* 1.队列名字
*/
channel.queueBind(queue,EXCHANGE_NAME,"");
System.out.println("wait print message...");
DeliverCallback deliverCallback = (var,var2)->{
System.out.println("01:"+new String(var2.getBody()));
};
channel.basicConsume(queue,true,deliverCallback, var->{});
}
}消费者02
package com.esint.rabbitmq.work03;
import com.esint.rabbitmq.RabbitMQUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
/**
* 消费者02的消息接受
*/
public class ReceiveLog02 {
public static final String EXCHANGE_NAME = "logs";
public static void main(String[] args) throws Exception {
//获取信道
Channel channel = RabbitMQUtils.getChannel();
//声明一个交换机
channel.exchangeDeclare(EXCHANGE_NAME,"fanout");
/**
*声明一个队列 队列名字随机
* 消费者断开与队列的链接后 队列自动删除
*/
String queue = channel.queueDeclare().getQueue();
/**
* 绑定交换机与队列
* 1.队列名字
*/
channel.queueBind(queue,EXCHANGE_NAME,"");
System.out.println("wait print message...");
DeliverCallback deliverCallback = (var,var2)->{
System.out.println("02:"+new String(var2.getBody(),"UTF-8"));
};
channel.basicConsume(queue,true,deliverCallback, var->{});
}
}两个临时队列：
运行测试生产者：
运行测试消费者：
