图文课程 职群旗下

  1. 首页
  2. 所有分类
  3. Java
  4. RabbitMQ 一二事(3) - 订阅模式(微信公众号模式)的应用

RabbitMQ 一二事(3) - 订阅模式(微信公众号模式)的应用

之前讲的消费者互相可以把队列中的消息全部读取,但是不是读完整的所有信息

那么采用订阅模式就行,这就是微信公众号的模式,

比如10个人订阅了我的公众号"BeJavaGod",当我发送一条消息的时候,

那么这10个人都能收到我的消息并且查看,比如本条消息,对吧?

生产者制造消息发送给交换机X,而不是发送给队列,队列和交换机绑定,消费者从各自的队列中获得消息

这样则实现一个生产者发送的所有消息都能被所有的消费者同时接收到

需要注意的地方是,在生产者创建消息发送到交换机时,此时没有队列,那么消息则丢失,消费者的队列绑定后再次发送则消息传达,原理是消息必须存放在队列中

生产者:


public class Send {

private final static String EXCHANGE_NAME = "test_exchange_fanout";


public static void main(String[] argv) throws Exception {

//
获取到连接以及mq通道
Connection connection = ConnectionUtil.getConnection();

Channel channel = connection.createChannel();



//
声明exchange
channel.exchangeDeclare(EXCHANGE_NAME, "fanout");



//
消息内容
String message = "id=1001";

channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());

System.out.println(" [x] Sent '" + message + "'");


channel.close();

connection.close();

}


}

消费者1


public class Recv {

private final static String QUEUE_NAME = "test_queue_fanout_1";


private final static String EXCHANGE_NAME = "test_exchange_fanout";


public static void main(String[] argv) throws Exception {


//
获取到连接以及mq通道
Connection connection = ConnectionUtil.getConnection();

Channel channel = connection.createChannel();



//
声明队列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);



//
绑定队列到交换机
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");



//
同一时刻服务器只会发一条消息给消费者
channel.basicQos(1);



//
定义队列的消费者
QueueingConsumer consumer = new QueueingConsumer(channel);


//
监听队列,手动返回完成
channel.basicConsume(QUEUE_NAME, false, consumer);



//
获取消息
while (true) {
QueueingConsumer.Delivery delivery = consumer.nextDelivery();

String message = new String(delivery.getBody());

System.out.println(" [x] Received '" + message + "'");

Thread.sleep(10);


channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);

}


}


}

消费者2


public class Recv2 {

private final static String QUEUE_NAME = "test_queue_fanout_2";


private final static String EXCHANGE_NAME = "test_exchange_fanout";


public static void main(String[] argv) throws Exception {


//
获取到连接以及mq通道
Connection connection = ConnectionUtil.getConnection();

Channel channel = connection.createChannel();



//
声明队列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);



//
绑定队列到交换机
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");



//
同一时刻服务器只会发一条消息给消费者
channel.basicQos(1);



//
定义队列的消费者
QueueingConsumer consumer = new QueueingConsumer(channel);


//
监听队列,手动返回完成
channel.basicConsume(QUEUE_NAME, false, consumer);



//
获取消息
while (true) {
QueueingConsumer.Delivery delivery = consumer.nextDelivery();

String message = new String(delivery.getBody());

System.out.println(" [x] Received '" + message + "'");

Thread.sleep(10);


channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);

}


}


}

ok,这样就是最简单的订阅demo

猜你喜欢
Java SE基础部分——常用类库之Math和Random类(随机产生数值) 使用Java实现简单串口通信
使用java传参调用exe并且获取程序进度和返回结果的一种方法 JSP学习笔记(4)-Javabean
RabbitMQ 一二事(4) - 路由模式介绍 RabbitMQ 一二事(5) - 通配符模式应用
用户线程和内核线程的区别 IOS开发基础知识--碎片40
android Volley+Gson的使用 jQuery分析(3) - jQuery.fn.init