rabbitmq系列(二)几种常见模式的应用场景及实现

一、简单模式

原理:生产者将消息交给默认的交换机,交换机获取消息后交给绑定这个生产者的队列(投递规则为队列名称和routing key 相同的队列),监听当前队列的消费者获取信息并执行消费逻辑。

rabbitmq系列(二)几种常见模式的应用场景及实现

场景:有一个oa系统,用户通过接收手机验证码进行注册,页面上点击获取验证码后,将验证码放到消息队列,然后短信服务从队列中获取到验证码,并发送给用户。

实现

生产者:

public class Producter {

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

        // 1. 创建出链接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(5672);
        factory.setVirtualHost("/");

        // 2. 通过链接工厂创建链接对象
        Connection connection = factory.newConnection();

        // 3. 通过链接对象创建出channel
        Channel channel = connection.createChannel();
        // 4. 通过channel发布消息
        /**
         * 四个参数:
         * 第一个参数是交换机的名称
         * 第二个参数是路由键
         * 第三个参数标识消息的一些额外的属性
         * 第四个是消息的具体的内容
         */
        String message = "字节";
        for(int i = 0;i < 5;i ++){

            channel.basicPublish("","byte001",null,message.getBytes());
        }
        // 5. 释放资源,释放channel 和 链接对象

        channel.close();
        connection.close();
    }
}

消费者:


public class Consumer {

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

        // 1. 创建出链接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(5672);
        factory.setVirtualHost("/");

        // 2. 通过链接工厂创建链接对象
        Connection connection = factory.newConnection();

        // 3. 通过链接对象创建出channel
        Channel channel = connection.createChannel();

        // 4. 创建出消息队列
        /**
         * 第一个参数是消息队列的名称
         * 第二个参数表示消息是否持久化
         * 第三个参数标识消息队列是否被channel独占
         * 第四个参数标识是否自动删除消息队列,当消息队列没有绑定交换机后是否自动删除
         * 第五个是消息队列扩展参数
         */
        String queueName = "byte001";
        channel.queueDeclare(queueName, true, false, false, null);

        // 5. 创建消费者,对消息进行处理
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            /**
             * consumerTag 用来标识.可以再监听队列时候设置
             * envelope 信封,通过envelope可以通过这个获取到很多东西
             * properties 额外的消息属性
             * body:消息体
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {

                String s = new String(body, "UTF-8");
                System.out.println("获取到的消息:"+s);
            }
        };
        // 6. 通过channel消费者和消息队列关联
        /**
         * 第一个参数是消息队列的名字
         * 第二个参数是否自动签收(即消费消息后告知服务器已被消费)
         * 第三个参数是消费者
         */
        channel.basicConsume(queueName, true, consumer);
    }
}

二、工作模式

原理:生产者将消息交给交换机,交换机交给绑定的队列,队列有多个消费者监听,一条消息只能由一个消费者消费,这样就形成了资源竞争,谁的资源空闲大,争抢到的可能性就大。

rabbitmq系列(二)几种常见模式的应用场景及实现

工作模式.png

场景:有一个电商平台,有两个订单服务,用户下单的时候,任意一个订单服务消费用户的下单请求生成订单即可。不用两个订单服务同时消费用户的下单请求。

实现

生产者:

public class Producter {

    public static final String QUEUE_NAME = "byte002";
    public static void main(String[] args) throws Exception {

        // 1. 创建出链接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(5672);
        factory.setVirtualHost("/");

        // 2. 通过链接工厂创建链接对象
        Connection connection = factory.newConnection();

        // 3. 通过链接对象创建出channel
        Channel channel = connection.createChannel();

        // 申明队列
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        // 4. 通过channel发布消息
        /**
         * 四个参数:
         * 第一个参数是交换机的名称
         * 第二个参数是路由键
         * 第三个参数标识消息的一些额外的属性
         * 第四个是消息的具体的内容
         */
        String message = "字节";
        for(int i = 0;i < 100;i ++){

            channel.basicPublish("",QUEUE_NAME,null,(message+i).getBytes());
        }
        // 5. 释放资源,释放channel 和 链接对象

        channel.close();
        connection.close();
    }
}

消费者1:


public class Consumer {

    public static final String QUEUE_NAME = "byte002";
    public static void main(String[] args) throws Exception {

        // 1. 创建出链接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(5672);
        factory.setVirtualHost("/");

        // 2. 通过链接工厂创建链接对象
        Connection connection = factory.newConnection();

        // 3. 通过链接对象创建出channel
        Channel channel = connection.createChannel();

        // 4. 创建出消息队列
        /**
         * 第一个参数是消息队列的名称
         * 第二个参数表示消息是否持久化
         * 第三个参数标识消息队列是否被channel独占
         * 第四个参数标识是否自动删除消息队列,当消息队列没有绑定交换机后是否自动删除
         * 第五个是消息队列扩展参数
         */
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        channel.basicQos(1); // 告诉服务器,在我们没有确认当前消息时不要给我们发送新的消息
        // 5. 创建消费者,对消息进行处理
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            /**
             * consumerTag 用来标识.可以再监听队列时候设置
             * envelope 信封,通过envelope可以通过这个获取到很多东西
             * properties 额外的消息属性
             * body:消息体
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {

                String s = new String(body, "UTF-8");
                System.out.println("消费者1收到的内容:"+s);
                try {
                    Thread.sleep(10); // 模拟消费耗时
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                channel.basicAck(envelope.getDeliveryTag(),false); // 参数2false为确认收到消息,true为拒绝收到消息
            }
        };
        // 6. 通过channel消费者和消息队列关联
        /**
         * 第一个参数是消息队列的名字
         * 第二个参数是否自动签收(即消费消息后告知服务器已被消费)
         * 第三个参数是消费者
         */
        channel.basicConsume(QUEUE_NAME, false, consumer);
    }
}

消费者2:

public class Consumer2 {

    public static final String QUEUE_NAME = "byte002";
    public static void main(String[] args) throws Exception {

        // 1. 创建出链接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(5672);
        factory.setVirtualHost("/");

        // 2. 通过链接工厂创建链接对象
        Connection connection = factory.newConnection();

        // 3. 通过链接对象创建出channel
        Channel channel = connection.createChannel();

        // 4. 创建出消息队列
        /**
         * 第一个参数是消息队列的名称
         * 第二个参数表示消息是否持久化
         * 第三个参数标识消息队列是否被channel独占
         * 第四个参数标识是否自动删除消息队列,当消息队列没有绑定交换机后是否自动删除
         * 第五个是消息队列扩展参数
         */
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        channel.basicQos(1); // 告诉服务器,在我们没有确认当前消息时不要给我们发送新的消息
        // 5. 创建消费者,对消息进行处理
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            /**
             * consumerTag 用来标识.可以再监听队列时候设置
             * envelope 信封,通过envelope可以通过这个获取到很多东西
             * properties 额外的消息属性
             * body:消息体
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {

                String s = new String(body, "UTF-8");
                System.out.println("消费者2收到的内容:"+s);
                try {
                    Thread.sleep(500); // 模拟消费耗时
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                channel.basicAck(envelope.getDeliveryTag(),false); // 参数2false为确认收到消息,true为拒绝收到消息
            }
        };
        // 6. 通过channel消费者和消息队列关联
        /**
         * 第一个参数是消息队列的名字
         * 第二个参数是否自动签收(即消费消息后告知服务器已被消费)
         * 第三个参数是消费者
         */
        channel.basicConsume(QUEUE_NAME, false, consumer);
    }
}

保证资源竞争的代码就是这一行channel.basicQos(1);如果不加这一行,我们会发现两个消费者是轮询消费消息的。

三、发布订阅模式

原理:生产者将消息扔给交换机,交换机类型是fanout,不同的队列注册到交换机上,不同的消费注册在不同的队列上。所有消费者都会收到消息。

rabbitmq系列(二)几种常见模式的应用场景及实现

场景:有一个商城,我们新添加一个商品后,可能同时需要去更新缓存和数据库。

实现:

生产者:

public class Producter {

    // 定义交换机的名字
    public static final String EXCHANGE_NAME="byte003";
    public static void main(String[] args) throws Exception {

        // 1. 创建出链接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(5672);
        factory.setVirtualHost("/");

        // 2. 通过链接工厂创建链接对象
        Connection connection = factory.newConnection();

        // 3. 通过链接对象创建出channel
        Channel channel = connection.createChannel();
        // 定义一个交换机,类型是fanout
        channel.exchangeDeclare(EXCHANGE_NAME,"fanout");

        // 因为消息先发到交换机,交换机没有保存功能,所以如果没有消费者,消息会丢失
        channel.basicPublish(EXCHANGE_NAME,"",null,"发布订阅模式的消息".getBytes());

        channel.close();
        connection.close();
    }
}

消费者1:

public class Consumer1 {
    public static final String EXCHANGE_NAME="byte003";

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


        // 1. 创建出链接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(5672);
        factory.setVirtualHost("/");

        // 2. 通过链接工厂创建链接对象
        Connection connection = factory.newConnection();

        // 3. 通过链接对象创建出channel
        Channel channel = connection.createChannel();

        String queueName = "queue003";
        channel.queueDeclare(queueName,false,false,false,null);

        // 绑定队列到交换机
        channel.queueBind(queueName,EXCHANGE_NAME,"");
        channel.basicQos(1);
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            /**
             * consumerTag 用来标识.可以再监听队列时候设置
             * envelope 信封,通过envelope可以通过这个获取到很多东西
             * properties 额外的消息属性
             * body:消息体
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {

                String s = new String(body, "UTF-8");
                System.out.println("消费者1:"+s);
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };
        channel.basicConsume(queueName,false,consumer);
    }
}

消费者2:

public class Consumer2 {
    public static final String EXCHANGE_NAME="byte003";

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


        // 1. 创建出链接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(5672);
        factory.setVirtualHost("/");

        // 2. 通过链接工厂创建链接对象
        Connection connection = factory.newConnection();

        // 3. 通过链接对象创建出channel
        Channel channel = connection.createChannel();

        String queueName = "queue004";
        channel.queueDeclare(queueName,false,false,false,null);

        // 绑定队列到交换机
        channel.queueBind(queueName,EXCHANGE_NAME,"");
        channel.basicQos(1);
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            /**
             * consumerTag 用来标识.可以再监听队列时候设置
             * envelope 信封,通过envelope可以通过这个获取到很多东西
             * properties 额外的消息属性
             * body:消息体
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {

                String s = new String(body, "UTF-8");
                System.out.println("消费者2:"+s);
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };
        channel.basicConsume(queueName,false,consumer);
    }
}

需要注意的一点就是交换机没有保存功能,如果没有消费者,则消息会丢失。

四、路由模式

原理:生产者将消息发送给交换机,消息携带具体的routingkey。交换机类型是direct,接收到消息中的routingkey,比对与之绑定的队列的routingkey,分发到不同的队列上。

rabbitmq系列(二)几种常见模式的应用场景及实现
rabbitmq系列(二)几种常见模式的应用场景及实现

场景:还是一样,有一个商城,新添加了一个商品,实时性不是很高,只需要添加到数据库即可,不用刷新缓存。

实现

生产者:

public class Producter {

    // 定义交换机的名字
    public static final String EXCHANGE_NAME="byte004";
    public static void main(String[] args) throws Exception {

        // 1. 创建出链接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(5672);
        factory.setVirtualHost("/");

        // 2. 通过链接工厂创建链接对象
        Connection connection = factory.newConnection();

        // 3. 通过链接对象创建出channel
        Channel channel = connection.createChannel();
        // 定义一个交换机,类型是direct
        channel.exchangeDeclare(EXCHANGE_NAME,"direct");

        // 因为消息先发到交换机,交换机没有保存功能,所以如果没有消费者,消息会丢失
        channel.basicPublish(EXCHANGE_NAME,"key1",null,"发布路由模式的消息".getBytes());

        channel.close();
        connection.close();
    }
}

消费者1:

public class Consumer1 {
    public static final String EXCHANGE_NAME="byte004";

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


        // 1. 创建出链接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(5672);
        factory.setVirtualHost("/");

        // 2. 通过链接工厂创建链接对象
        Connection connection = factory.newConnection();

        // 3. 通过链接对象创建出channel
        Channel channel = connection.createChannel();

        String queueName = "queue005";
        channel.queueDeclare(queueName,false,false,false,null);

        // 绑定队列到交换机
        /**
         * 参数3是routingkey,只有和它一样的routingkey的消息才会被当前消费者收到
         */
        channel.queueBind(queueName,EXCHANGE_NAME,"key1");
        // 如果要接收多个routingkey的消息,在执行一次上面的代码即可,如下
        channel.queueBind(queueName,EXCHANGE_NAME,"key2");
        channel.basicQos(1);
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            /**
             * consumerTag 用来标识.可以再监听队列时候设置
             * envelope 信封,通过envelope可以通过这个获取到很多东西
             * properties 额外的消息属性
             * body:消息体
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {

                String s = new String(body, "UTF-8");
                System.out.println("消费者1:"+s);
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };
        channel.basicConsume(queueName,false,consumer);
    }
}

消费者2:


public class Consumer2 {
    public static final String EXCHANGE_NAME="byte004";

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


        // 1. 创建出链接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(5672);
        factory.setVirtualHost("/");

        // 2. 通过链接工厂创建链接对象
        Connection connection = factory.newConnection();

        // 3. 通过链接对象创建出channel
        Channel channel = connection.createChannel();

        String queueName = "queue006";
        channel.queueDeclare(queueName,false,false,false,null);

        // 绑定队列到交换机
        channel.queueBind(queueName,EXCHANGE_NAME,"key2");
        channel.basicQos(1);
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            /**
             * consumerTag 用来标识.可以再监听队列时候设置
             * envelope 信封,通过envelope可以通过这个获取到很多东西
             * properties 额外的消息属性
             * body:消息体
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {

                String s = new String(body, "UTF-8");
                System.out.println("消费者2:"+s);
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };
        channel.basicConsume(queueName,false,consumer);
    }
}

五、主题模式

原理:路由模式的一种,路由功能添加了模糊匹配。星号(*)代表1个单词,#号(#)代表一个或多个单词。具体可参考路由模式。

rabbitmq系列(二)几种常见模式的应用场景及实现

主题模式.png

场景:还是一样,有一个商城,新添加了一个商品,实时性不是很高,只需要添加到数据库即可,数据库包含了主数据库mysql1和从数据库mysql2的内容,不用刷新缓存。

实现

生产者:


public class Producter {

    // 定义交换机的名字
    public static final String EXCHANGE_NAME="byte004";
    public static void main(String[] args) throws Exception {

        // 1. 创建出链接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(5672);
        factory.setVirtualHost("/");

        // 2. 通过链接工厂创建链接对象
        Connection connection = factory.newConnection();

        // 3. 通过链接对象创建出channel
        Channel channel = connection.createChannel();
        // 定义一个交换机,类型是topic
        channel.exchangeDeclare(EXCHANGE_NAME,"topic");

        // 因为消息先发到交换机,交换机没有保存功能,所以如果没有消费者,消息会丢失
        channel.basicPublish(EXCHANGE_NAME,"key.1.2",null,"发布路由模式的消息".getBytes());

        channel.close();
        connection.close();
    }
}

消费者1:

public class Consumer1 {
    public static final String EXCHANGE_NAME="byte004";

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


        // 1. 创建出链接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(5672);
        factory.setVirtualHost("/");

        // 2. 通过链接工厂创建链接对象
        Connection connection = factory.newConnection();

        // 3. 通过链接对象创建出channel
        Channel channel = connection.createChannel();

        String queueName = "queue005";
        channel.queueDeclare(queueName,false,false,false,null);

        // 绑定队列到交换机
        /**
         * 参数3是routingkey,只有和它一样的routingkey的消息才会被当前消费者收到
         */
        channel.queueBind(queueName,EXCHANGE_NAME,"key.*");
        // 如果要接收多个routingkey的消息,在执行一次上面的代码即可,如下
        channel.queueBind(queueName,EXCHANGE_NAME,"abc.#");
        channel.basicQos(1);
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            /**
             * consumerTag 用来标识.可以再监听队列时候设置
             * envelope 信封,通过envelope可以通过这个获取到很多东西
             * properties 额外的消息属性
             * body:消息体
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {

                String s = new String(body, "UTF-8");
                System.out.println("消费者1:"+s);
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };
        channel.basicConsume(queueName,false,consumer);
    }
}

消费者2:

public class Consumer2 {
    public static final String EXCHANGE_NAME="byte004";

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


        // 1. 创建出链接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(5672);
        factory.setVirtualHost("/");

        // 2. 通过链接工厂创建链接对象
        Connection connection = factory.newConnection();

        // 3. 通过链接对象创建出channel
        Channel channel = connection.createChannel();

        String queueName = "queue006";
        channel.queueDeclare(queueName,false,false,false,null);

        // 绑定队列到交换机
        channel.queueBind(queueName,EXCHANGE_NAME,"key.#");
        channel.basicQos(1);
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            /**
             * consumerTag 用来标识.可以再监听队列时候设置
             * envelope 信封,通过envelope可以通过这个获取到很多东西
             * properties 额外的消息属性
             * body:消息体
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {

                String s = new String(body, "UTF-8");
                System.out.println("消费者2:"+s);
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        };
        channel.basicConsume(queueName,false,consumer);
    }
}

代码已上传:

github地址: https://github.com/binzh303/zhixie-code-example

gitee地址: https://gitee.com/javaXiaoCaiJi/zhixie-code-example

原文出处:微信公众号【 字节传说 】

原文链接:https://mp.weixin.qq.com/s?__biz=MzU1MzczMzExNA==&mid=2247483983&idx=1&sn=8960904c90918d008d86d55713a8074f&chksm=fbef1cdacc9895cc73e7eaf8aab121af9dd46767c7667c0135ca50b95300ddb6fc804facf6ef&mpshare=1&scene=1&srcid=&sharer_sharetime=1578874623202&sharer_shareid=cc609ea2d56165426b8c4ccd3a28d316&key=24f4c229624f482365642745a2b5f5392becea1a41bdc9d7d10ccd6b5b743cfcdbe346ef140582de32e4b11ecf204a332c808c13fc1fac887ef4b396a773fe42622c65f17d35b15bf1b5d28cc0240293&ascene=1&uin=Mjk4MzIyNjQxOQ%3D%3D&devicetype=Windows+10&version=6208006a&lang=zh_CN&exportkey=A32mD9iLAdrLRQab5ZfjDO0%3D&pass_ticket=y%2BH%2FKAJlOLTOrFV8oDmS8nZ3D%2BHntzrCm6B9SGS1aPlsJ9V%2BtUjAnlVlcA6dKWbN

本文观点不代表Dotnet9立场,转载请联系原作者。

发表评论

登录后才能评论