解决RocketMQ的幂等性问题_Java

造成重复消费的原因

当系统的调用链路比较长的时候，比如系统a调用系统b，系统b再把消息发送到rocketmq中，在系统a调用系统b的时候，如果系统b处理成功，但是迟迟没有将调用成功的结果返回给系统a的时候，系统a就会尝试重新发起请求给系统b，造成系统b重复处理，发起多条消息给rocketmq造成重复消费；
在系统b发送给rocketmq的时候，也有可能会发生和上面一样的问题，消息发送超时，结果系统b重试，导致rocketmq接收到了重读消息；
当rocketmq成功接收到消息，并将消息交给消费者处理，如果消费者消费完成后还没来得及提交consume_success给rocketmq，自己宕机或者重启了，那么rocketmq没有接收到consume_success，就会认为消费失败了，会重发消息给消费者再次消费；

通过幂等性来保证，只要保证重复消息不对结果产生影响，就完美地解决这个问题。

解决方法

生产者端

rocketmq支持消息查询的功能，只要去rocketmq查询一下是否已经发送过该条消息就可以了，不存在则发送，存在则不发送，也就是message.setkeys()；
引入redis，在发送消息到rocketmq成功之后，向redis中插入一条数据，如果发送重试，则先去redis查询一个该条消息是否已经发送过了，存在的话就不重复发送消息了；

缺点
方法一：rocketmq消息查询的性能不是特别好，如果在高并发的场景下，每条消息在发送到rocketmq时都去查询一下，可能会影响接口的性能；

方法二：在一些极端的场景下，redis也无法保证消息发送成功之后，就一定能写入redis成功，比如写入消息成功而redis此时宕机，那么再次查询redis判断消息是否已经发送过，是无法得到正确结果的；

消费者端

建立一个消息表，拿到这个消息做数据库的insert操作。给这个消息做一个唯一主键（primary key）或者唯一约束，那么就算出现重复消费的情况，就会导致主键冲突。
拿到这个消息做redis的set的操作.redis就是天然幂等性

代码实现

方式一：
生产者

public class mqproducer {
    public static void main(string[] args) throws mqclientexception {
        //创建生产者
        defaultmqproducer producer=new defaultmqproducer("rmq-group");
        //设置nameserver地址
        producer.setnamesrvaddr("192.168.138.187:9876;192.168.138.188:9876");
        //设置生产者实例名称
        producer.setinstancename("producer");
        //启动生产者
        producer.start();
        try {
            //发送消息
            for (int i=0;i<1;i++){
                thread.sleep(1000); //每秒发送一次
                //创建消息
                message msg = new message("wn04", // topic 主题名称
                        "taga", // tag 临时值
                        ("w-"+i).getbytes()// body 内容
                );
                //消息的唯一標識
                msg.setkeys(system.currenttimemillis() + "");
                //发送消息
                sendresult sendresult=producer.send(msg);
                system.out.println(sendresult.tostring());
            }
        } catch (exception e) {
            e.printstacktrace();
        }
        producer.shutdown();
    }

}

消费者端：

public class mqconsumer {

    //保存标识的集合
    static private map<string, string> logmap = new hashmap<>();

    public static void main(string[] args) throws mqclientexception {
        //创建消费者
        defaultmqpushconsumer consumer=new defaultmqpushconsumer("rmq-group");
        //设置nameserver地址
        consumer.setnamesrvaddr("192.168.138.187:9876;192.168.138.188:9876");
        //设置消费者实例名称
        consumer.setinstancename("consumer");
        //订阅topic
        consumer.subscribe("wn04","taga");
        //监听消息
        consumer.registermessagelistener(new messagelistenerconcurrently() {
            @override
            public consumeconcurrentlystatus consumemessage(list<messageext> list, consumeconcurrentlycontext consumeconcurrentlycontext) {
                string key = null;
                string msgid = null;
                try {
                    for (messageext msg : list) {
                        key = msg.getkeys();
                        //判断集合当中有没有存在key,存在就不需要重试,不存在先存key再回来重试后消费消息
                        if (logmap.containskey(key)) {
                            // 无需继续重试。
                            system.out.println("key:"+key+",无需重试...");
                            return consumeconcurrentlystatus.consume_success;
                        }
                        msgid = msg.getmsgid();
                        system.out.println("key:" + key + ",msgid:" + msgid + "---" + new string(msg.getbody()));
                        //模拟异常
                        int i = 1 / 0;
                    }
                } catch (exception e) {
                    //e.printstacktrace();
                    //重试
                    return consumeconcurrentlystatus.reconsume_later;
                } finally {
                    //保存key
                    logmap.put(key, msgid);
                }

                return consumeconcurrentlystatus.consume_success;
            }
        });
        consumer.start();
        system.out.println("consumer started...");

    }
}

到此这篇关于解决rocketmq的幂等性问题的文章就介绍到这了,更多相关rocketmq 幂等性内容请搜索代码网以前的文章或继续浏览下面的相关文章希望大家以后多多支持代码网！