rabbitmq 工作模式一

痛定思痛。 2022-04-13 07:22 211阅读 0赞

学习工作模式前,先看一下rabbitmq 给的helloworld案例

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70][]

这是传统的一对一,,,,  也就是一台机器生产,一台机器接收....

为了更好的了解代码....我这里演示的话用底层的代码来演示....不整合框架了

<dependency>
        <groupId>com.rabbitmq</groupId>
        <artifactId>amqp-client</artifactId>
        <version>4.0.3</version>
    </dependency>

这是rabbitmq 提供的依赖......导入一下就可以测试了

下面是我写的生产类...

测试是没问题的

import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    public class Producer01 {
    
        //队列名称
        private static final String QUEUE = "helloworld";
    
        public static void main(String[] args) throws IOException, TimeoutException {
    
    
    
    
            Connection connection = null;   //与
            Channel channel = null;
    
    
            try {
    
                ConnectionFactory factory = new ConnectionFactory();
    
                //设置ip
                factory.setHost("localhost");
    
                //设置端口
                factory.setPort(5672);
    
                //设置账号密码
                factory.setUsername("guest");  //默认账号密码都是guest
                factory.setPassword("guest");
    
                //设置虚拟空间
                factory.setVirtualHost("/");//虚拟机默认的虚拟名称为/ , 虚拟机相当于一个独立的服务器
    
                //创建与RabbitMQ服务的TCP连接
                connection = factory.newConnection();
    
                //创建连接通道 ,每个连接可以创建多个通道,每个通道只有一个会话
                channel = connection.createChannel();
    
                //这有五个参数
                /***
                 * 声明队列，如果Rabbit中没有此队列将自动创建*
                 * param1:队列名称*
                 * param2:是否持久化*  rabbit 关闭了该队列是否存在..
                 * param3:队列是否独占此连接*    如果参数是true,那么一个连接connection 只能存在这一个channel,除非关闭程序
                 * param4:队列不再使用时是否自动删除此队列*    该队列不使用了就会删除该队列
                 *
                 * param5:队列参数*/
                channel.queueDeclare(QUEUE, true, false, false,null );
    
                String message = "你爱到极致的人,不会爱你";
    
    
                //发布消息
                /**
                 * String exchange, String routingKey,  BasicProperties props, byte[] body
                 *
                 * param1  : 交换机 后面我会讲这里是指定交换机,使用默认的交换机
                 * param2  :  路由key,这也是先不写,,后面讲   大概作用是用于Exchange（交换机）将消息转发到指定的消息队列
                 *  param3 :消息包含的属性
                 *   消息体
                 *
                 */
                channel.basicPublish("", QUEUE,null ,message.getBytes() );
    
                System.out.println("Send Message is:'" + message + "'");
            }catch (Exception e){
    
            }finally {
                if (channel != null){
                    channel.close();
                }
                
                if (connection != null){
                    connection.close();
                }
            }
        }
    }

然后下面是我写的消费类 ,,  连接mq代码都一样来着...关注发送消息和接收消息的方法就行 ...

同样测试过,代码是可运行的

import com.rabbitmq.client.*;
    
    import java.io.IOException;
    
    public class Consumer01 {
    
        //队列名称
        private static final String QUEUE = "helloworld";
    
        public static void main(String[] args){
    
            Connection connection = null;
            Channel channel = null;
    
            try {
    
    
    
                ConnectionFactory factory = new ConnectionFactory();
    
                //设置ip
                factory.setHost("localhost");
    
                //设置端口
                factory.setPort(5672);
    
                //设置账号密码
                factory.setUsername("guest");  //默认账号密码都是guest
                factory.setPassword("guest");
    
                //设置虚拟空间
                factory.setVirtualHost("/");//虚拟机默认的虚拟名称为/ , 虚拟机相当于一个独立的服务器
    
                //创建与RabbitMQ服务的TCP连接
                connection = factory.newConnection();
    
                //创建连接通道 ,每个连接可以创建多个通道,每个通道只有一个会话
                channel = connection.createChannel();
    
                //这有五个参数
                /***
                 * 声明队列，如果Rabbit中没有此队列将自动创建*
                 * param1:队列名称*
                 * param2:是否持久化*  rabbit 关闭了该队列是否存在..
                 * param3:队列是否独占此连接*    如果参数是true,那么一个连接connection 只能存在这一个channel,除非关闭程序
                 * param4:队列不再使用时是否自动删除此队列*    该队列不使用了就会删除该队列
                 *
                 * param5:队列参数*/
                channel.queueDeclare(QUEUE, true, false, false,null );  //这里其实可以不用声明队列的,因为 生产者已经声明过了,但是如果生产者后发布服务,队列没有声明,消费者去监听队列..会报错
    
    
    //            创建默认消费方法
                DefaultConsumer consumer = new DefaultConsumer(channel) {
    
    //                重写监听方法
    
    
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
    
    
                        System.out.println("receive message.."+new String(body,"utf-8"));
                    }
                };
    
    
    //            监听队列
                /**
                 * String queue, boolean autoAck, Consumer callback
                 *
                 * param1 :  队列名称
                 * param2 :   是否自动回复,接收到消息会自动恢复mq收到了,mq会删除消息,如果拒绝的话需要手动回复,不回复的话会导致mq不删除被消费过的消息,一直存在
                 * param3 : 消费对象,,包含消费方法
                 *
                 */
                channel.basicConsume(QUEUE,true , consumer);
            }catch (Exception e){
    
            } 
        }
    }

以上的话就是rabbitmq提供的案例

一台生产者 , 一台消费者

emmm,,,

这里讲的工作的模式是 workqueues

## WorkQueues ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70 1][]

对比helloword案例,这里多了个消费者..

应用场景：对于 任务过重或任务较多情况使用工作队列可以提高任务处理的速度。

测试

我们启动俩次消费者

然后用刚刚写的生产者,发送五条信息

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70 2][]

然后我们看消费者打印的信息

![20181128165640763.png][]

![20181128165652785.png][]

结果 :

mq workqueues 使用的是轮询方式讲信息平均发给消费者 ,

消费者会在处理完消息后 接收下一条消息

##  2.Publish/subscribe 发布订阅模式 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70 3][]

特点

生产者将消息发送给broker.由交换机将消息发给每个跟绑定了交换机绑定的消息队列,每个队列都能收到生产者发送的每一条消息

生产者 :

声明Exchange\_fanout\_inform交换机。

声明两个队列并且绑定到此交换机，

绑定时不需要指定routingkey发送消息时不需要指定routingkey

import com.rabbitmq.client.BuiltinExchangeType;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    import java.io.IOException;
    
    public class Producer02 {
    
        //消息队列名称
        public static final String QUEUE_INFORM_Test1 = "queue_inform_1";
        public static final String QUEUE_INFORM_Test2 = "queue_inform_2";
    
        //交换机名称
        public static final String EXCHANGE_FANOUT_INFORM="exchange_fanout_inform";
    
    
        public static void main(String[] args){
    
            Connection connection = null;
            Channel channel = null;
    
    
            try {
    
                ConnectionFactory factory = new ConnectionFactory();
    
                //设置ip
                factory.setHost("localhost");
    
                //设置端口
                factory.setPort(5672);
    
                //设置账号密码
                factory.setUsername("guest");  //默认账号密码都是guest
                factory.setPassword("guest");
    
                //设置虚拟空间
                factory.setVirtualHost("/");//虚拟机默认的虚拟名称为/ , 虚拟机相当于一个独立的服务器
    
                //创建与RabbitMQ服务的TCP连接
                connection = factory.newConnection();
    
                //创建连接通道 ,每个连接可以创建多个通道,每个通道只有一个会话
                channel = connection.createChannel();
    
    
                //声明交换机
    
                /*
                    String exchange, String type
                    param1 :  交换机
                    param2 : 交换机 类型   fanout 、 topic、direct、headers
    
                    FANOUT 对应的模式是 发布订阅模式  publish/subscribe 模式
    
                    其他的工作模式以后会将
                    DIRECT  对应的是路由的工作模式
                    TOPIC  对应的是通配符工作模式
                    HEADERS   对应了 headers 的工作模式
                 */
                channel.exchangeDeclare(EXCHANGE_FANOUT_INFORM,BuiltinExchangeType.FANOUT);
    
    
                //声明队列
    
                /**
                 * String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
                 *
                 * param1: 队列名称
                 * param2: 是否持久化
                 * param3 : 是否独占此队列
                 * param4 : 队列不用是否自动删除
                 *  param5 : 参数
                 */
                channel.queueDeclare(QUEUE_INFORM_Test1,true ,false ,false ,null );
                channel.queueDeclare(QUEUE_INFORM_Test2,true ,false ,false ,null );
    
    
                //交换机和队列绑定
                /**
                 * String queue, String exchange, String routingKey
                 * param1 :  队列名称
                 * exchange :  交换机
                 * routingKey : 路由key  后面讲,先 用 ""代替
                 */
                channel.queueBind(QUEUE_INFORM_Test1,EXCHANGE_FANOUT_INFORM ,"" );
                channel.queueBind(QUEUE_INFORM_Test2,EXCHANGE_FANOUT_INFORM ,"" );
    
    
    //            发送消息
                String message = "";
                for (int i = 0; i < 9; i++) {
                    message = "故事的开头总是这样，适逢其会，猝不及防。故事的结局总是这样，花开两朵，天各一方。"+ i;
    
    
                    /**
                     * String exchange, String routingKey, BasicProperties props, byte[] body
                     *
                     *   param1  交换机名称
                     *   param2 路由key,后面讲,先用 "" 代替 ,
                     *   param3  参数
                     *   param4 传递的字符串
                     *
                     *
                     */
                    channel.basicPublish(EXCHANGE_FANOUT_INFORM,"" , null, message.getBytes());
    
                    System.out.println("Send Message is:'" + message + "'");
                }
            }catch (Exception e){
    
            }finally{
                if(channel!=null){
                    try {
                        channel.close();
                    } catch (Exception e) {
                        e.printStackTrace();
                    } 
                }
                if(connection!=null){
                    try {
                        connection.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
    
    
    
    
    
    
    
            }
    }

下面是我写的俩个消费者

import com.rabbitmq.client.*;
    
    import java.io.IOException;
    
    public class ConsumerTest1 {
    
        //队列名称
        private static final String QUEUE_INFORM_Test1 = "queue_inform_1";
        //交换机名称
        public static final String EXCHANGE_FANOUT_INFORM="exchange_fanout_inform";
    
        public static void main(String[] args){
    
            Connection connection = null;
            Channel channel = null;
    
            try {
    
    
    
                ConnectionFactory factory = new ConnectionFactory();
    
                //设置ip
                factory.setHost("localhost");
    
                //设置端口
                factory.setPort(5672);
    
                //设置账号密码
                factory.setUsername("guest");  //默认账号密码都是guest
                factory.setPassword("guest");
    
                //设置虚拟空间
                factory.setVirtualHost("/");//虚拟机默认的虚拟名称为/ , 虚拟机相当于一个独立的服务器
    
                //创建与RabbitMQ服务的TCP连接
                connection = factory.newConnection();
    
                //创建连接通道 ,每个连接可以创建多个通道,每个通道只有一个会话
                channel = connection.createChannel();
    
                //这有五个参数
                /***
                 * 声明队列，如果Rabbit中没有此队列将自动创建*
                 * param1:队列名称*
                 * param2:是否持久化*  rabbit 关闭了该队列是否存在..
                 * param3:队列是否独占此连接*    如果参数是true,那么一个连接connection 只能存在这一个channel,除非关闭程序
                 * param4:队列不再使用时是否自动删除此队列*    该队列不使用了就会删除该队列
                 *
                 * param5:队列参数*/
                channel.queueDeclare(QUEUE_INFORM_Test1, true, false, false,null );  //这里其实可以不用声明队列的,因为 生产者已经声明过了,但是如果生产者后发布服务,队列没有声明,消费者去监听队列..会报错
    
    
    
    
                //声明交换机
    
                /*
                    String exchange, String type
                    param1 :  交换机
                    param2 : 交换机 类型   fanout 、 topic、direct、headers
    
                    FANOUT 对应的模式是 发布订阅模式  publish/subscribe 模式
    
                    其他的工作模式以后会将
                    DIRECT  对应的是路由的工作模式
                    TOPIC  对应的是通配符工作模式
                    HEADERS   对应了 headers 的工作模式
                 */
                channel.exchangeDeclare(EXCHANGE_FANOUT_INFORM,BuiltinExchangeType.FANOUT);
    
                //交换机和队列绑定
                /**
                 * String queue, String exchange, String routingKey
                 * param1 :  队列名称
                 * exchange :  交换机
                 * routingKey : 路由key  后面讲,先 用 ""代替
                 */
                channel.queueBind(QUEUE_INFORM_Test1,EXCHANGE_FANOUT_INFORM ,"" );
    
    
    
    
    //            创建默认消费方法
                DefaultConsumer consumer = new DefaultConsumer(channel) {
    
    //                重写监听方法
    
    
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
    
    
                        System.out.println("receive message.."+new String(body,"utf-8"));
                    }
                };
    
    
    //            监听队列
                /**
                 * String queue, boolean autoAck, Consumer callback
                 *
                 * param1 :  队列名称
                 * param2 :   是否自动回复,接收到消息会自动恢复mq收到了,mq会删除消息,如果拒绝的话需要手动回复,不回复的话会导致mq不删除被消费过的消息,一直存在
                 * param3 : 消费对象,,包含消费方法
                 *
                 */
                channel.basicConsume(QUEUE_INFORM_Test1,true , consumer);
            }catch (Exception e){
    
            }
        }
    }

消费者二

import com.rabbitmq.client.ConnectionFactory;
    import com.rabbitmq.client.DefaultConsumer;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    
    public class ConsumerTest2 {
    
        //队列名称
    
        public static final String QUEUE_INFORM_Test2 = "queue_inform_2";
    
        //交换机名称
        public static final String EXCHANGE_FANOUT_INFORM="exchange_fanout_inform";
    
        public static void main(String[] args){
    
            Connection connection = null;
            Channel channel = null;
    
            try {
    
    
    
                ConnectionFactory factory = new ConnectionFactory();
    
                //设置ip
                factory.setHost("localhost");
    
                //设置端口
                factory.setPort(5672);
    
                //设置账号密码
                factory.setUsername("guest");  //默认账号密码都是guest
                factory.setPassword("guest");
    
                //设置虚拟空间
                factory.setVirtualHost("/");//虚拟机默认的虚拟名称为/ , 虚拟机相当于一个独立的服务器
    
                //创建与RabbitMQ服务的TCP连接
                connection = factory.newConnection();
    
                //创建连接通道 ,每个连接可以创建多个通道,每个通道只有一个会话
                channel = connection.createChannel();
    
                //这有五个参数
                /***
                 * 声明队列，如果Rabbit中没有此队列将自动创建*
                 * param1:队列名称*
                 * param2:是否持久化*  rabbit 关闭了该队列是否存在..
                 * param3:队列是否独占此连接*    如果参数是true,那么一个连接connection 只能存在这一个channel,除非关闭程序
                 * param4:队列不再使用时是否自动删除此队列*    该队列不使用了就会删除该队列
                 *
                 * param5:队列参数*/
                channel.queueDeclare(QUEUE_INFORM_Test2, true, false, false,null );  //这里其实可以不用声明队列的,因为 生产者已经声明过了,但是如果生产者后发布服务,队列没有声明,消费者去监听队列..会报错
    
    
    
    
                //声明交换机
    
                /*
                    String exchange, String type
                    param1 :  交换机
                    param2 : 交换机 类型   fanout 、 topic、direct、headers
    
                    FANOUT 对应的模式是 发布订阅模式  publish/subscribe 模式
    
                    其他的工作模式以后会将
                    DIRECT  对应的是路由的工作模式
                    TOPIC  对应的是通配符工作模式
                    HEADERS   对应了 headers 的工作模式
                 */
                channel.exchangeDeclare(EXCHANGE_FANOUT_INFORM,BuiltinExchangeType.FANOUT);
    
    
    
    
    
                //交换机和队列绑定
                /**
                 * String queue, String exchange, String routingKey
                 * param1 :  队列名称
                 * exchange :  交换机
                 * routingKey : 路由key  后面讲,先 用 ""代替
                 */
                channel.queueBind(QUEUE_INFORM_Test2,EXCHANGE_FANOUT_INFORM ,"" );
    
    
    
    
    //            创建默认消费方法
                DefaultConsumer consumer = new DefaultConsumer(channel) {
    
    //                重写监听方法
    
    
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
    
    
                        System.out.println("receive message.."+new String(body,"utf-8"));
                    }
                };
    
    
    //            监听队列
                /**
                 * String queue, boolean autoAck, Consumer callback
                 *
                 * param1 :  队列名称
                 * param2 :   是否自动回复,接收到消息会自动恢复mq收到了,mq会删除消息,如果拒绝的话需要手动回复,不回复的话会导致mq不删除被消费过的消息,一直存在
                 * param3 : 消费对象,,包含消费方法
                 *
                 */
                channel.basicConsume(QUEUE_INFORM_Test2,true , consumer);
            }catch (Exception e){
    
            }
        }
    }

细心的同学可能发现了,,生产者跟消费者的不同代码其实就是发送消息,接收消息的方法而已 ....

生产者声明队列跟交换机,消费者也声明各自的队列跟交换机...  其实就是为了怕先启动消费者没有发现队列跟交换机报错而已...

核心代码的话  其实  就是 生成交换机,生成队列 绑定交换机跟队列

测试的话

我们先启动俩个消费者

![20181128194303258.png][]

然后我们启动生产者

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70 4][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70 5][]

我们看打印结果,,俩台消费者各自都处理了9条信息

其实这种方法比WORKQUEUES 工作模式强,因为多台机器可以监听一个队列,也就是下图所示,我们可以要俩个队列,当然也可以创建一个队列...创建多少队列跟一个队列多少消费者完全取决与我们

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70 6][]

因为时间问题,,,,还有4种工作模式下次写- -

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70]: /images/20220413/19f8918f00d1443ba00e80352972dde9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70 1]: /images/20220413/8bab67cd35c2457a9a83b48679afdb58.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70 2]: /images/20220413/c99aa8885cb84a7995cf1efe0c4cbff3.png
[20181128165640763.png]: /images/20220413/fb2b30d2adcb48948054d444a0ce8a4e.png
[20181128165652785.png]: /images/20220413/58f5c10792f3465a9526b0cd35af4785.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70 3]: /images/20220413/0161df9594314e278d92a92e705b3ba5.png
[20181128194303258.png]: /images/20220413/0334ebccfe294747a891d183416a25b4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70 4]: /images/20220413/41133405868349c0b847912c5acf27d3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70 5]: /images/20220413/7ad72e8d53734af984ee133f1898df6c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwNzk0MjY2_size_16_color_FFFFFF_t_70 6]: /images/20220413/d1a6376e4f5b49388e86cefde20a49b5.png