扇出模式 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->{});}
}
两个临时队列:
运行测试生产者:
运行测试消费者: