25 | 异步处理好用，但非常容易用错

我之前做过一个demo 是基于canal做mysql数据同步，需要将解析好的数据发到kafka里面，再进行处理。在使用的时候发现这么一个问题，就是kafka多partition消费时不能保证消息的顺序消费，进而导致mysql数据同步异常。 由于kafka可以保证在同一个partition内消息有序，于是我自定义了一个分区器，将数据的id取hashcode然后根据partition的数量取余作为分区号，保证同一条数据的binlog能投递到同一个partition中，从而达到消息顺序消费的目的。

老师，我理解的异步处理不仅仅是通过 MQ 来实现，还有其他方式 比如开新线程执行，返回 Future 还有各种异步框架，比如 Vertx，它是通过 callback 的方式实现 思考题 1.可能是数据写到了主库，然后查询了从库。但因为主从同步有延迟，导致没有查询到

第一个问题： 每天晒白牙大佬的回答和老师的回复已经很棒了，我就不班门弄斧了。 第二个问题： 自定义的私信队列，其实是发送失败，主要是生产者发送到mq的时候，发送失败，进了自定义的私信队列； DLX的方式的方式其实解决已到了mq，但是因为各种原因，无法到达正常的队列中，大概分类下面几种吧： 消息消费时被拒绝(basic.reject / basic.nack)，并且requeue = false 消息TTL过期 队列达到最大长度 分享一下之前在公司内部分享的RabbitMQ的资料，欢迎大家交流 github上传不上去，只能用有道云笔记，请大家见谅 资料主要从： MQ选型及特点； AMQP与RabbitMQ模型； RabbitMQ核心概念； RabbitMQ相关机制； 这几个点去分析的，请大家多多指教。 http://note.youdao.com/noteshare?id=e9f2f88c6c7fcb7ac690463eb230650a

老师 我这里有个问题 关于Stream的，业务需求里需要按某几个字段去重(acctId,billingCycleId,prodInstId,offerId) 我这里想到了遍历集合areaDatas 后用contains方法判断 重写AcctItemYzfBean实体类的equals方法实现， 请问有没有更好的方法？ 代码如下 List<AcctItemYzfBean> newList = new CopyOnWriteArrayList<>(); //循环过滤、增强翼支付数据 Optional.ofNullable(areaDatas)//集合判空 .orElse(new ArrayList<>()) .stream()//转化为流 便于下面过滤和增强数据 .filter(Objects::nonNull)//元素判空 .filter(yzfBean -> this.judgeIfOfferId(yzfBean))//判断销售品ID是否相同 .filter(yzfBean -> this.enhanceYzfBean(yzfBean))//增强过滤accNbr和acctId .filter(yzfBean -> this.judgeIfArrears(yzfBean))//判断是否不欠费 .filter(yzfBean -> this.judgeIfCancel(yzfBean))//判断是否销账金额大于0 .filter(yzfBean -> this.judgeIfReturn(yzfBean))//判断是否上月未返还 .forEach(yzfBean -> { //去重 重写AcctItemYzfBean.equals方法 if(!newList.contains(yzfBean)) { //增强latnName yzfBean.setLatnName(commonRegionMap.get(yzfBean.getRegionId())); //增强areaCode yzfBean.setAreaCode(areaCode); //数据封装 newList.add(yzfBean); } }); 重写的equals方法 @Override public boolean equals(Object yzfBeanObj) { if(yzfBeanObj instanceof AcctItemYzfBean) { AcctItemYzfBean yzfBean = (AcctItemYzfBean) yzfBeanObj; if(Tools.isEmpty(yzfBean.getAcctId(), yzfBean.getBillingCycleId(), yzfBean.getProdInstId(), yzfBean.getOfferId())) { return false; } if(yzfBean.getAcctId().equals(this.acctId) && yzfBean.getBillingCycleId().equals(this.billingCycleId) && yzfBean.getProdInstId().equals(this.prodInstId) && yzfBean.getOfferId().equals(this.offerId)) { return true; } } return super.equals(yzfBeanObj); }

老师 1.如果实际生产中用使用 ConcurrentHashMap 来存放那些发过短信的用户 ID 实现幂等，如何一直往map中增加，会不会oom呢？ 2.如果数据量巨大 使用ConcurrentSkipListMap 跳表会不会更好一些呢？

mq发信息写到了事务中，导致了mq的消费时，事务还没有提交

老师，你好，如果有多个补偿实例，会不会造成消息重复？

在event sourcing的时候，有一种方案是outbox，在发送方维持一个数据表，这样可以保证消息和业务数据在一个事务中，也是一种消息发送记录的方式

第一个问题，是否可以这样解决 1.先保存用户注册的数据，同时记录下要发送mq的消息，入库在一个事务里 2.通过异步任务定时拉取mq的消息表，发送到mq，进行处理 但这个有个问题，异步任务就能执行mq的的业务，那mq的价值是不是减少了

对了 许多同时用@Async标签实现 希望能够做到，但是又不用100%保证的解耦动作。实际也有隐患，不放把@Async标签底层实现更换为mq，是不是更合适？