RabbitMQ(一)之工作模式、生产者消息确认机制

1、work queues 工作队列模式

工作队列模式是多个消费者同时消费同一个队列中的数据,默认情况下,MQ会把所有请求平均发送给每一个消费者。可以设置处理完成后在获取下一条。channel.basicQos(1),消费者伪代码如下:

        //获取TCP长连接
        Connection conn = RabbitUtils.getConnection();
        //创建通信“通道”,相当于TCP中的虚拟连接
        Channel channel = conn.createChannel();

        //创建队列,声明并创建一个队列,如果队列已存在,则使用这个队列
        //第一个参数:队列名称ID
        //第二个参数:是否持久化,false对应不持久化数据,MQ停掉数据就会丢失
        //第三个参数:是否队列私有化,false则代表所有消费者都可以访问,true代表只有第一次拥有它的消费者才能一直使用,其他消费者不让访问
        //第四个:是否自动删除,false代表连接停掉后不自动删除掉这个队列
        //其他额外的参数, null
        channel.queueDeclare("rabbit-sms",false, false, false, null);
        //如果不写basicQos(1),则自动MQ会将所有请求平均发送给所有消费者
        //basicQos,MQ不再对消费者一次发送多个请求,而是消费者处理完一个消息后(确认后),在从队列中获取一个新的
        channel.basicQos(1);//处理完一个取一个
        channel.basicConsume(RabbitConstant.QUEUE_SMS , false , new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String jsonSMS = new String(body);
                System.out.println("短信发送成功:" + jsonSMS);
                //false只确认签收当前的消息,设置为true的时候则代表签收该消费者所有未签收的消息
                channel.basicAck(envelope.getDeliveryTag() , false);
            }
        });

2、Pub、Sub 发布订阅模式

后面的模式会使用到交换机(Exchange),交换机是接收生产者发送的消息,在根据Exchange类型不同区别发送到消费者,或者将消息丢弃,Exchange只负责转发消息,不具备存储消息的能力。因此如果没有任何队列与Exchange绑定,或者没有符合路由规则的队列,那么消息会丢失

Exchange类型:

Fanout:广播,将消息发送给所有绑定到交换机的队列
Direct:定向,把消息发送给符合指定routingkey的队列
Topic:通配符,把消息发送给符合routing pattern(路由模式)的队列

发布订阅模式是所有消费者获得的消息相同。与工作队列模式不同的是,生产者和消费者需要绑定到同一个交换机上。工作队列模式不需要设置交换机(实际上,底层也会实用一个默认的交换机)。

生产者伪代码:

        Connection connection = RabbitUtils.getConnection();
        Channel channel = connection.createChannel();
        channel.basicPublish("交换机名称","" , null , "需要发送的内容".getBytes());
        channel.close();
        connection.close();

消费者伪代码:

        Connection connection = RabbitUtils.getConnection();
        final Channel channel = connection.createChannel();
        channel.queueDeclare(RabbitConstant.QUEUE_BAIDU, false, false, false, null);

        //queueBind用于将队列与交换机绑定
        channel.queueBind("自己的队列名", "交换机名称", "路由key(暂时用不到)");
        channel.basicQos(1);
        channel.basicConsume(RabbitConstant.QUEUE_BAIDU , false , new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("订阅到的信息:" + new String(body));
                channel.basicAck(envelope.getDeliveryTag() , false);
            }
        });

3、Routing 路由模式

队列与交换机的绑定需要制定一个RoutingKey(路由的key),消息发送方在向交换机(Exchange)发送消息时,也必须指定RoutingKey,Exchange不在把消息发送给每一个绑定的队列,而是根据消息的RoutingKey进行判断,只有绑定队列的RoutingKey与生产消息的RoutingKey完全一致才会收到消息。RoutingKey可以绑定多个。

生产者伪代码:

        //省略了创建连接和关闭连接的代码
        channel.basicPublish("交换机的名称","所要发送到哪个RoutingKey(RoutingKey1、RoutingKey2)" , null , "需要发送的内容");

消费者伪代码:

        //省略了创建连接的代码
        channel.queueDeclare("队列名称", false, false, false, null);
        //queueBind用于将队列与交换机绑定
        //参数1:队列名 参数2:交互机名  参数三:路由key
        channel.queueBind("队列名称", "Exchange  Name", "RoutingKey1");
        channel.queueBind("队列名称", "Exchange  Name", "RoutingKey2");
        channel.basicQos(1);
        channel.basicConsume("队列名称" , false , new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("接受到的信息:" + new String(body));
                channel.basicAck(envelope.getDeliveryTag() , false);
            }
        });

4、Topic 通配符模式

Topic模式与Routing模式工作类似,Topic模式的Exchange可以让队列在绑定RoutingKey的时候使用通配符,RoutingKey一般是用一个或者多个单词组成,例如:order.type1.pay、order.type2.pay。

通配符规则: #表示匹配一个或者多个词,只能匹配一个词。例如:order.#可以匹配上述两个key,order.就不能匹配到,order.type1.*就可以匹配到order.type1.pay。

Topic模式下的发布者与Routing模式下的相同。只是消费者有所区别

消费者伪代码如下:

//省略了创建链接的代码
//RourtingKey可以使用通配符进行匹配。order.#可以匹配到order.type1.pay、order.type2.pay,多个路由
channel.queueBind("队列名称", "Exchange Name", "order.#");

5、RabbitMQ的消息确认机制

RabbitMQ提供了一个监听器(Listener)来接收消息的投递状态,消息确认涉及到两种状态,Confirm和Return:

Confirm代表生产者将消息送到Broker时产生的状态,包括:ack代表Broker已经将数据接收成功,nack代表Broker拒收消息。原因可能有很多种,如队列已满、限流、IO异常等等。

Return代表消息已经被Broker正常接收,消息已经ack,但Broker没有对应的队列进行投递时,产生的状态,消息将被退回给生产者。

上述两种状态只代表生产者与Broker之间的消息投递情况,与消费者是否接收和确认无关

        //省略了创建连接的代码
        //开启confirm监听模式
        channel.confirmSelect();
        //Confirm
        channel.addConfirmListener(new ConfirmListener() {
            public void handleAck(long l, boolean b) throws IOException {
                //第二个参数代表接收的数据是否为批量接收,一般我们用不到。
                System.out.println("消息已被Broker接收,Tag:" + l );
            }
            public void handleNack(long l, boolean b) throws IOException {
                System.out.println("消息已被Broker拒收,Tag:" + l);
            }
        });
        //Return
       channel.addReturnListener(new ReturnCallback() {
            public void handle(Return r) {
                System.err.println("===========================");
                System.err.println("Return编码:" + r.getReplyCode() + "-Return描述:" + r.getReplyText());
                System.err.println("交换机:" + r.getExchange() + "-路由key:" + r.getRoutingKey() );
                System.err.println("Return主题:" + new String(r.getBody()));
                System.err.println("===========================");
            }
        });
        //次处省略了发送消息的相关代码
最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
【社区内容提示】社区部分内容疑似由AI辅助生成,浏览时请结合常识与多方信息审慎甄别。
平台声明:文章内容(如有图片或视频亦包括在内)由作者上传并发布,文章内容仅代表作者本人观点,简书系信息发布平台,仅提供信息存储服务。

相关阅读更多精彩内容

友情链接更多精彩内容