Rabbitmq confirm 异步模式

Posted mm163

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Rabbitmq confirm 异步模式相关的知识,希望对你有一定的参考价值。

//存储未确认的消息标识tag
final SortedSet<Long> confirmSet = Collections.synchronizedNavigableSet(new TreeSet<Long>());

增加监听器
channel.addConfirmListener(new ConfirmListener() {


/*
* 消息确认返回成功
* l:如果是多条,这个就是最后一条消息的tag
* b:是否多条
* */
@Override
public void handleAck(long l, boolean b) throws IOException {
System.out.println("消息发送成功"+l+"是否多条"+b);
if(b){
confirmSet.headSet(l+1).clear();
}else{
confirmSet.remove(l);
}
}

/*消息确认返回失败*/
@Override
public void handleNack(long l, boolean b) throws IOException {
System.out.println("消息发送失败"+l+"是否多条"+b);
if(b){
confirmSet.headSet(l+1).clear();
}else{
confirmSet.remove(l);
}
}
});

String routingKey ="goods.delete";
for (int i = 0; i <10; i++) {
String message = "hello ps"+i;
long tag = channel.getNextPublishSeqNo();
channel.basicPublish(Exchange_NAME,routingKey,null,message.getBytes("utf-8"));
System.out.println(tag);
confirmSet.add(tag);
}
发送者
package com.aynu.bootamqp.service;

import com.aynu.bootamqp.commons.utils.Amqp;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;


import java.io.IOException;
import java.util.Collections;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.TimeoutException;

public class Send {

    private final static String Exchange_NAME ="hello";
    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
            Connection connection = Amqp.getConnection();
            Channel channel = connection.createChannel();
            //声明交换机
            channel.exchangeDeclare(Exchange_NAME,"topic");
            //在手动确认机制之前
            //一次只发送一条消息,给不同的消费者
            channel.basicQos(1);
            //将通道设置为comfirm模式
            channel.confirmSelect();

        //存储未确认的消息标识tag
        final SortedSet<Long> confirmSet = Collections.synchronizedNavigableSet(new TreeSet<Long>());

        channel.addConfirmListener(new ConfirmListener() {

            /*
             * 消息确认返回成功
             * l:如果是多条,这个就是最后一条消息的tag
             * b:是否多条
             * */
            @Override
            public void handleAck(long l, boolean b) throws IOException {
                System.out.println("消息发送成功"+l+"是否多条"+b);
                if(b){
                    confirmSet.headSet(l+1).clear();
                }else{
                    confirmSet.remove(l);
                }
            }

            /*消息确认返回失败*/
            @Override
            public void handleNack(long l, boolean b) throws IOException {
                System.out.println("消息发送失败"+l+"是否多条"+b);
                if(b){
                    confirmSet.headSet(l+1).clear();
                }else{
                    confirmSet.remove(l);
                }
            }
        });


            String routingKey ="goods.delete";
            for (int i = 0; i <10; i++) {
                String message = "hello ps"+i;
                long tag = channel.getNextPublishSeqNo();
                channel.basicPublish(Exchange_NAME,routingKey,null,message.getBytes("utf-8"));
                System.out.println(tag);
                confirmSet.add(tag);
            }


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

接受者

package com.aynu.bootamqp.service;

import com.aynu.bootamqp.commons.utils.Amqp;
import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;
@SuppressWarnings("all")
public class Receive2 {

    private final static String QUEUE_NAME ="hello1";
    private final static String Exchange_NAME ="hello";
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = Amqp.getConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        channel.queueBind(QUEUE_NAME,Exchange_NAME,"goods.#");
        channel.basicQos(1);
        DefaultConsumer consumer = new DefaultConsumer(channel) {

            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body) throws IOException {
                super.handleDelivery(consumerTag, envelope, properties, body);
                String msg = new String(body,"utf-8");
                System.out.println(msg);
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }finally {
                    // 手动发送消息确认机制
                    channel.basicAck(envelope.getDeliveryTag(),false);
                }
            }
        };
        boolean autoAck = false;
        channel.basicConsume(QUEUE_NAME,autoAck,consumer);
    }
}

 

 

以上是关于Rabbitmq confirm 异步模式的主要内容,如果未能解决你的问题,请参考以下文章

消息队列-一篇读懂rabbitmq(生命周期,confirm模式,延迟队列,集群)

RabbitMQ - 发布确认

13.RabbitMQ 消息可靠性投递confirm确认模式

RabbitMQ事物模式

二万字长文图文详解RabbitMQ6 种工作模式(理论与代码相结合)

RabbitMq-confirm发送消息确认深入探讨