借鉴AQS的CHL思路解决消息多线程消费顺序ACK问题
借鉴AQS的CHL思路解决消息多线程消费顺序ACK问题
背景
我们的支付场景下，要求消费的业务消息绝不能丢失，且能充分利用高规格的服务器的性能，比如用线程池对业务消息进行快速处理。有同学可能没太理解这个问题有啥不好处理，让我一步步分析下。
MQ的优势和缺点
MQ是我们在应对高并发场景最常用的一种措施，它可以帮我们对业务解耦、对流程异步化以及削峰填谷的妙用。
但是，由于引入了这一额外的中间件，也增加了系统的复杂度和不稳定因素。
消息可靠性的应对
消息的可靠性保证需要从消息流转的每个环节进行保障，比如生产端的事务型消息，broker的实时刷盘持久化，消费端的手动ACK 。
这里，我们对生产端和存储端的保障措施不作讨论，重点关注消费端的手动ACK机制。
手动ACK的问题
手动ACK可以保证消息一定被消费，但是需要确保手动ACK的顺序和消息顺序一致，为什么？
消息队列之所以性能高处理快，是因为采用了文件顺序读写方式，系统在拉取消息进行消费时，是按顺序文件的offset进行拉取的，如果commit offset的顺序错乱，会使得服务端的消息状态错乱，比如消息重发。
因此，如果我们在本地启动了线程池，对消息进行拉取处理，由于各线程的处理速度不一定一致，所以无法保证各线程处理完之后对各自消息的ACK操作是顺序的，怎么办，难道只能同步拉消费取然后ACK么。
解决方案
最不济，可以提交一批任务，批量等待统一提交。不过总觉得不优雅。
某次看JUC中的AQS的时候，启发了我。
我们平时用的类似CountDownLauch这些并发工具类，不也是处理的多线程协作的问题么。
我们的场景完全没有AQS复杂，借鉴它的思路，应该是没有问题的。
- 创建双端队列，队列节点中需要维护自身处理状态state，和对应msg的offset。
- 服务从消息中心拉取消息，在提交本地线程池执行之前，先入队列。
- 消息消费完之后，通知队列中对应的节点，更新状态为完成。
- 队列头被更新后出队列，提交offset，并判断新的队列头的状态，直到遇到state是未完成的head时阻塞。 undefined
方案解析
该方案可以有效利用本地线程的资源，并行的处理，并通过队列和异步通知机制保证最终commit offset时有序。
在最差情况下(即head节点对应的msg最后一个被处理完)，相当于等待一批线程处理完成后统一提交。除此之外等待性能都要更优。
异步通知的实现
public class MSGFuture {
/*全局变量，存放msg对应的future对象*/
private static final Map<Long, MSGFuture> FUTURES = new ConcurrentHashMap<Long, MSGFuture>();
/*全局不变唯一标识*/
private final long id;
/*最长等待时间*/
private final int timeout;
/*并发锁*/
private final Lock lock = new ReentrantLock();
/*通知条件*/
private final Condition done = lock.newCondition();
/*开始时间*/
private final long start = System.currentTimeMillis();
/*业务结果*/
private volatile Object response;
}
//构造函数
public MSGFuture(Request request, int timeout) {
/*全局自增ID*/
this.id = request.getrId();
/*超时时间*/
this.timeout = timeout > 0 ? timeout : 1000;
/*放入全局变量*/
FUTURES.put(id, this);
}
//业务处理结果更新
public static void received(long id, Object response) {
MSGFuture future = FUTURES.remove(id);
if (future != null) {
future.doReceived(response);
} else {
logger.warn("response return timeout,id:"+id);
}
}
//结果更新，通知等待条件
private void doReceived(Object res) {
lock.lock();
try {
response = res;
done.signal();
} finally {
lock.unlock();
}
}
//异步等待获取结果
public Object get(int timeout) throws TimeoutException {
if (!isDone()) {
long start = System.currentTimeMillis();
lock.lock();
try {
while (!isDone()) {
done.await(timeout, TimeUnit.MILLISECONDS);
if (isDone() || System.currentTimeMillis() - start > timeout) {
break;
}
}
} catch (InterruptedException e) {
throw new RuntimeException(e);
} finally {
lock.unlock();
}
if (!isDone()) {
throw new TimeoutException();
}
}
return returnFromResponse();
}
总结
看到这里，有同学会说，这个和AQS有啥关系呀~
其实，只是处理思路的一种借鉴，比如state状态，比如锁机制和通知等待。既然都是多线程任务协调，那总有相似之处。
总之一句话，别说背八股文没用，多多了解会有大帮助~