RabbitMQ简单应用の订阅模式

订阅模式

公众号-->订阅之后才会收到相应的文章。

解读：
1.一个生产者，多个消费者
2.每个消费者都有自己的队列
3.生产者没有将消息直接发送到队列里，而是发送给了交换机（转发器）exchange
4.每个队列都要绑定到交换机（转发器）上
5.生产者发送的消息记过交换机然后到达队列，然后就能实现被多个消费者消费

图例：

　　　　　|-------------|-----Q-----C3

P------------X-------------|-----Q-----C3

　　　　　|-------------|-----Q-----C3

MQ工厂类Connection

 package com.mmr.rabbitmq.util;

 import java.io.IOException;

 import com.rabbitmq.client.Connection;

 import com.rabbitmq.client.ConnectionFactory;

 public class ConnectionUtils {

     /**

      * @desc 获取Mq 的链接

      * @author zp

      * @throws IOException

      * @date 2018-7-19

      */

     public static  Connection getConnection() throws IOException {

         // 1.定义一个链接工厂

         ConnectionFactory factroy = new ConnectionFactory();

         // 2.设置服务地址

         factroy.setHost("127.0.0.1");

         // 3.设置端口号

         factroy.setPort(5672);

         // 4.vhost  设置数据库

         factroy.setVirtualHost("vhtest");

         // 5.设置用户名

         factroy.setUsername("jerry");

         // 6. 设置密码

         factroy.setPassword("123456");

         // 7.返回链接

         return factroy.newConnection();

     }

 }

消息生产者类Send，这个时候，运行代码再到控制台去查看，并没有发现我们的消息，因为在MQ中只有队列可以存储消息，而交换机不可以存储消息，下面这段代码并没有将交换机和队列进行绑定，所以数据就丢失了。

 package com.mmr.rabbitmq.ps;

 import java.io.IOException;

 import com.mmr.rabbitmq.util.ConnectionUtils;

 import com.rabbitmq.client.Channel;

 import com.rabbitmq.client.Connection;

 public class Send {

     private static final String EXCHANGE_NAME="test_exchange_fanout";

     public static void main(String[] args) throws IOException {

         // 创建连接

         Connection connection = ConnectionUtils.getConnection();

         // 获取通道

         Channel channel = connection.createChannel();

         // 声明交换机

         channel.exchangeDeclare(EXCHANGE_NAME, "fanout");// fanout 分发

         // 发送消息

         String msg = "hello ps";

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

         System.out.println("send:"+msg);

         channel.close();

         connection.close();

     }

 }

代码运行后的控制台：

由于交换机不能存储数据，那么我们就需要考虑如何将交换机和队列进行绑定。因为只要将两者进行绑定之后，那么数据存储问题就迎刃而解。

消费者Recv1 Recv2

 package com.mmr.rabbitmq.ps;

 import java.io.IOException;

 import com.mmr.rabbitmq.util.ConnectionUtils;

 import com.rabbitmq.client.Channel;

 import com.rabbitmq.client.Connection;

 import com.rabbitmq.client.Consumer;

 import com.rabbitmq.client.DefaultConsumer;

 import com.rabbitmq.client.Envelope;

 import com.rabbitmq.client.AMQP.BasicProperties;

 public class Recv1 {

     private static final String QUEUE_NAME_STRING="test_queue_fanout_email";

     private static final String EXCHANGE_NAME="test_exchange_fanout";

     public static void main(String[] args) throws IOException {

         // 创建连接

         Connection connection = ConnectionUtils.getConnection();

         // 创建通道

         final Channel channel = connection.createChannel();

         // 声明队列

         channel.queueDeclare(QUEUE_NAME_STRING, false, false, false, null);

         // 绑定队列，绑定到交换机/转发器

         channel.queueBind(QUEUE_NAME_STRING, EXCHANGE_NAME, "");

         // 保证每次只分发一个

         channel.basicQos(1);

         // 定义一个消费者

         Consumer consumer = new DefaultConsumer(channel){

             @Override

             public void handleDelivery(String consumerTag, Envelope envelope,

                     BasicProperties properties, byte[] body) throws IOException {

                 // TODO Auto-generated method stub

                 String msg = new String(body,"utf-8");

                 System.out.println("[1]Recv msg:"+msg);

                 try {

                     // 每次休息一会儿

                     Thread.sleep(2000);

                 } catch (Exception e) {

                     // TODO: handle exception

                     e.printStackTrace();

                 }finally{

                     System.out.println("recv1 done");

                     //回执

                     channel.basicAck(envelope.getDeliveryTag(), false);

                 }

             }

         };

         boolean autoAck = false;// 不自动应答

         channel.basicConsume(QUEUE_NAME_STRING, autoAck,consumer);

     }

 }

 package com.mmr.rabbitmq.ps;

 import java.io.IOException;

 import com.mmr.rabbitmq.util.ConnectionUtils;

 import com.rabbitmq.client.Channel;

 import com.rabbitmq.client.Connection;

 import com.rabbitmq.client.Consumer;

 import com.rabbitmq.client.DefaultConsumer;

 import com.rabbitmq.client.Envelope;

 import com.rabbitmq.client.AMQP.BasicProperties;

 public class Recv2 {

     private static final String QUEUE_NAME_STRING="test_queue_fanout_sms";

     private static final String EXCHANGE_NAME="test_exchange_fanout";

     public static void main(String[] args) throws IOException {

         // 创建连接

         Connection connection = ConnectionUtils.getConnection();

         // 创建通道

         final Channel channel = connection.createChannel();

         // 声明队列

         channel.queueDeclare(QUEUE_NAME_STRING, false, false, false, null);

         // 绑定队列，绑定到交换机/转发器

         channel.queueBind(QUEUE_NAME_STRING, EXCHANGE_NAME, "");

         // 保证每次只分发一个

         channel.basicQos(1);

         // 定义一个消费者

         Consumer consumer = new DefaultConsumer(channel){

             @Override

             public void handleDelivery(String consumerTag, Envelope envelope,

                     BasicProperties properties, byte[] body) throws IOException {

                 // TODO Auto-generated method stub

                 String msg = new String(body,"utf-8");

                 System.out.println("[2]Recv msg:"+msg);

                 try {

                     // 每次休息一会儿

                     Thread.sleep(2000);

                 } catch (Exception e) {

                     // TODO: handle exception

                     e.printStackTrace();

                 }finally{

                     System.out.println("recv2 done");

                     //回执

                     channel.basicAck(envelope.getDeliveryTag(), false);

                 }

             }

         };

         boolean autoAck = false;// 不自动应答

         channel.basicConsume(QUEUE_NAME_STRING, autoAck,consumer);

     }

 }

运行上述代码进行监听，再通过运行Send发送消息，我们可以在MQ-管理平台上看到：

进过这样的使用，我们的消息订阅就完成了。

巴特西

RabbitMQ简单应用の订阅模式

最新文章

热门文章