RabbitMq入门（二）常用的工作模式

蔚落 2022-02-14 11:57 189阅读 0赞

# 0、["Hello World!"][Hello World] #

![python-one.png][]

代码 请参考 [代码][Link 1]

# 1、 [Work queues][] #

![python-two.png][]

工作队列模式相比于helloworld入门程序就是同一个队列被多个消费者绑定，

结果：  
1、一条消息只会被一个消费者接收；  
2、rabbit采用轮询的方式将消息是平均发送给消费者的；  
3、消费者在处理完某条消息后，才会收到下一条消息。

# 2、[Publish/Subscribe][Publish_Subscribe]  #

![python-three.png][]

消息生成者代码

package com.whwe.rabbitmqdemo.produc;
    
    import com.rabbitmq.client.BuiltinExchangeType;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /**
     * @ClassName Producer02
     * @Description TODO
     * @Author yueyiming
     * @Date 2019/4/26 11:49
     * @Version 1.0
     * https://blog.csdn.net/hzau_itdog
     **/
    public class Producer02 {
        //声明两个队列1个交换机
        private static final String QUEUE_INFORM_EMAIL = "queue_inform_email";
        private static final String QUEUE_INFORM_SMS = "queue_inform_sms";
        private static final String EXCHANGE_FANOUT_INFORM = "exchange_fanout_inform";
    
        public static void main(String[] args) throws IOException, TimeoutException {
            Connection connection = null;
            Channel channel = null;
            //创建连接
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("192.168.20.129");
            connectionFactory.setPort(5672);
            connectionFactory.setUsername("guest");
            connectionFactory.setPassword("guest");
            connectionFactory.setVirtualHost("/");
            try {
                connection = connectionFactory.newConnection();
                //创建channel通道
                channel = connection.createChannel();
                //申明交换机
                //类型  DIRECT("direct"), FANOUT("fanout"), TOPIC("topic"), HEADERS("headers");
                channel.exchangeDeclare(EXCHANGE_FANOUT_INFORM, BuiltinExchangeType.FANOUT);
    
                //声明队列
                channel.queueDeclare(QUEUE_INFORM_EMAIL,true,false,true,null);
                channel.queueDeclare(QUEUE_INFORM_SMS,true,false,true,null);
    
                //队列和交换机进行绑定
                //queue 队列名称
                //exchange 交换机名称
                //routingKey 路由key
                channel.queueBind(QUEUE_INFORM_EMAIL, EXCHANGE_FANOUT_INFORM, "");
                channel.queueBind(QUEUE_INFORM_SMS, EXCHANGE_FANOUT_INFORM, "");
                //发送消息
                for (int i = 0; i < 100; i++) {
                    StringBuilder sb = new StringBuilder("inform to user---");
                    sb.append(i);
                    channel.basicPublish(EXCHANGE_FANOUT_INFORM, "", null, sb.toString().getBytes("utf-8"));
                    System.out.println("发送消息------" + sb.toString());
                }
    
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                if (channel != null) {
                    channel.close();
                }
                if (connection != null) {
                    connection.close();
                }
            }
        }
    
    }

消息消费者

package com.whwe.rabbitmqdemo.conc;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    
    /**
     * @ClassName Consumer02_subscribe_email
     * @Description TODO
     * @Author yueyiming
     * @Date 2019/4/26 14:35
     * @Version 1.0
     * https://blog.csdn.net/hzau_itdog
     **/
    public class Consumer02_subscribe_email {
        private static final String QUEUE_INFORM_EMAIL = "queue_inform_email";
        private static final String EXCHANGE_FANOUT_INFORM = "exchange_fanout_inform";
    
        public static void main(String[] args) {
            Connection connection = null;
            Channel channel = null;
            //创建连接
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("192.168.20.129");
            connectionFactory.setPort(5672);
            connectionFactory.setUsername("guest");
            connectionFactory.setPassword("guest");
            connectionFactory.setVirtualHost("/");
            try {
                connection = connectionFactory.newConnection();
                //创建channel通道
                channel = connection.createChannel();
                //申明交换机
                //类型  DIRECT("direct"), FANOUT("fanout"), TOPIC("topic"), HEADERS("headers");
                channel.exchangeDeclare(EXCHANGE_FANOUT_INFORM, BuiltinExchangeType.FANOUT);
    
                //声明队列
                channel.queueDeclare(QUEUE_INFORM_EMAIL, true, false, true, null);
    
                //队列和交换机进行绑定
                //queue 队列名称
                //exchange 交换机名称
                //routingKey 路由key
                channel.queueBind(QUEUE_INFORM_EMAIL, EXCHANGE_FANOUT_INFORM, "");
    
                //定义消费方法
                DefaultConsumer consumer = new DefaultConsumer(channel) {
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                        //消息id
                        long msgId = envelope.getDeliveryTag();
                        //交换机名称
                        String exchange = envelope.getExchange();
                        //路由key
                        String routingKey = envelope.getRoutingKey();
                        //是否重新传递
                        boolean redeliver = envelope.isRedeliver();
                        System.out.println("msgId------" + msgId);
                        System.out.println("exchange------" + exchange);
                        System.out.println("routingKey------" + routingKey);
                        System.out.println("redeliver------" + redeliver);
                        System.out.println("接受消息为：------" + new String(body, "utf-8"));
    
                    }
                };
                channel.basicConsume(QUEUE_INFORM_EMAIL,true,consumer);
            } catch (Exception e) {
                e.printStackTrace();
            }
    
        }
    }

# 3、[Routing][] #

![python-four.png][]

消息生产者

package com.whwe.rabbitmqdemo.produc;
    
    import com.rabbitmq.client.BuiltinExchangeType;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    /**
     * @ClassName Producer03_routing
     * @Description TODO
     * @Author yueyiming
     * @Date 2019/4/26 14:59
     * @Version 1.0
     * https://blog.csdn.net/hzau_itdog
     **/
    public class Producer03_routing {
        //声明两个队列1个交换机
        private static final String QUEUE_INFORM_EMAIL = "queue_inform_email";
        private static final String QUEUE_INFORM_SMS = "queue_inform_sms";
        private static final String EXCHANGE_DIRECT_INFORM = "exchange_direct_inform";
    
        public static void main(String[] args) throws IOException, TimeoutException {
            Connection connection = null;
            Channel channel = null;
            //创建连接
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("192.168.20.129");
            connectionFactory.setPort(5672);
            connectionFactory.setUsername("guest");
            connectionFactory.setPassword("guest");
            connectionFactory.setVirtualHost("/");
            try {
                connection = connectionFactory.newConnection();
                //创建channel通道
                channel = connection.createChannel();
                //申明交换机
                //类型  DIRECT("direct"), FANOUT("fanout"), TOPIC("topic"), HEADERS("headers");
                channel.exchangeDeclare(EXCHANGE_DIRECT_INFORM, BuiltinExchangeType.DIRECT);
    
                //声明队列
                channel.queueDeclare(QUEUE_INFORM_EMAIL, true, false, true, null);
                channel.queueDeclare(QUEUE_INFORM_SMS, true, false, true, null);
    
                //队列和交换机进行绑定
                //queue 队列名称
                //exchange 交换机名称
                //routingKey 路由key
                channel.queueBind(QUEUE_INFORM_EMAIL, EXCHANGE_DIRECT_INFORM, "0");
                channel.queueBind(QUEUE_INFORM_SMS, EXCHANGE_DIRECT_INFORM, "1");
                //发送消息
                for (int i = 0; i < 100; i++) {
                    StringBuilder sb = new StringBuilder("inform to user---");
                    sb.append(i);
                    channel.basicPublish(EXCHANGE_DIRECT_INFORM, "" + i % 2, null, sb.toString().getBytes("utf-8"));
                    System.out.println("发送消息------" + sb.toString());
                }
    
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                if (channel != null) {
                    channel.close();
                }
                if (connection != null) {
                    connection.close();
                }
            }
        }
    }

消息消费者

package com.whwe.rabbitmqdemo.conc;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    
    /**
     * @ClassName Consumer03_routing_email
     * @Description TODO
     * @Author yueyiming
     * @Date 2019/4/26 15:04
     * @Version 1.0
     * https://blog.csdn.net/hzau_itdog
     **/
    public class Consumer03_routing_email {
        private static final String QUEUE_INFORM_EMAIL = "queue_inform_email";
        private static final String EXCHANGE_DIRECT_INFORM = "exchange_direct_inform";
    
        public static void main(String[] args) {
            Connection connection = null;
            Channel channel = null;
            //创建连接
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("192.168.20.129");
            connectionFactory.setPort(5672);
            connectionFactory.setUsername("guest");
            connectionFactory.setPassword("guest");
            connectionFactory.setVirtualHost("/");
            try {
                connection = connectionFactory.newConnection();
                //创建channel通道
                channel = connection.createChannel();
                //申明交换机
                //类型  DIRECT("direct"), FANOUT("fanout"), TOPIC("topic"), HEADERS("headers");
                channel.exchangeDeclare(EXCHANGE_DIRECT_INFORM, BuiltinExchangeType.DIRECT);
    
                //声明队列
                channel.queueDeclare(QUEUE_INFORM_EMAIL, true, false, true, null);
    
                //队列和交换机进行绑定
                //queue 队列名称
                //exchange 交换机名称
                //routingKey 路由key
                channel.queueBind(QUEUE_INFORM_EMAIL, EXCHANGE_DIRECT_INFORM, "0");
    
                //定义消费方法
                DefaultConsumer consumer = new DefaultConsumer(channel) {
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                        //消息id
                        long msgId = envelope.getDeliveryTag();
                        //交换机名称
                        String exchange = envelope.getExchange();
                        //路由key
                        String routingKey = envelope.getRoutingKey();
                        //是否重新传递
                        boolean redeliver = envelope.isRedeliver();
                        System.out.println("msgId------" + msgId);
                        System.out.println("exchange------" + exchange);
                        System.out.println("routingKey------" + routingKey);
                        System.out.println("redeliver------" + redeliver);
                        System.out.println("接受消息为：------" + new String(body, "utf-8"));
    
                    }
                };
                channel.basicConsume(QUEUE_INFORM_EMAIL, true, consumer);
            } catch (Exception e) {
                e.printStackTrace();
            }
    
        }
    }

Routing模式和Publish/subscibe有啥区别？

Routing模式要求队列在绑定交换机时要指定routingkey，消息会转发到符合routingkey的队列。

# 4、[Topics][] #

![python-five.png][]

Topics与[Routing][]唯一不同的是路由key中包含匹配符，功能比[Routing][]更加强大

队列绑定交换机指定通配符：  
统配符规则：  
中间以“.”分隔。  
符号\#可以匹配多个词，

符号\*可以匹配一个词语。

# 5、 [RPC][] #

![python-six.png][]

RPC即客户端远程调用服务端的方法 ，使用MQ可以实现RPC的异步调用，基于Direct交换机实现，流程如下：  
1、客户端即是生产者就是消费者，向RPC请求队列发送RPC调用消息，同时监听RPC响应队列。  
2、服务端监听RPC请求队列的消息，收到消息后执行服务端的方法，得到方法返回的结果  
3、服务端将RPC方法 的结果发送到RPC响应队列  
4、客户端（RPC调用方）监听RPC响应队列，接收到RPC调用结果。

[Hello World]: https://www.rabbitmq.com/tutorials/tutorial-one-python.html
[python-one.png]: /images/20220214/4573b10d3e39480caeaac76eda904b8e.png
[Link 1]: https://blog.csdn.net/hzau_itdog/article/details/89465906
[Work queues]: https://www.rabbitmq.com/tutorials/tutorial-two-python.html
[python-two.png]: /images/20220214/068a91dd1006466d8607532562897029.png
[Publish_Subscribe]: https://www.rabbitmq.com/tutorials/tutorial-three-python.html
[python-three.png]: /images/20220214/122f9c39490b48f9991e793a9154e5c6.png
[Routing]: https://www.rabbitmq.com/tutorials/tutorial-four-python.html
[python-four.png]: /images/20220214/4b12617235034c508e8edfccbf473691.png
[Topics]: https://www.rabbitmq.com/tutorials/tutorial-five-python.html
[python-five.png]: /images/20220214/a11615b3052b4deb8225336e2b4993c2.png
[RPC]: https://www.rabbitmq.com/tutorials/tutorial-six-python.html
[python-six.png]: /images/20220214/a482a43508ef41bfaa4996b53cf0c156.png