Springboot异步消息处理的方法

这篇文章主要介绍“Springboot异步消息处理的方法”的相关知识，小编通过实际案例向大家展示操作过程，操作方法简单快捷，实用性强，希望这篇“Springboot异步消息处理的方法”文章能帮助大家解决问题。

成都创新互联公司主要从事网页设计、PC网站建设（电脑版网站建设）、wap网站建设（手机版网站建设）、成都响应式网站建设、程序开发、网站优化、微网站、重庆小程序开发等，凭借多年来在互联网的打拼，我们在互联网网站建设行业积累了丰富的成都网站制作、成都网站建设、网站设计、网络营销经验，集策划、开发、设计、营销、管理等多方位专业化运作于一体。

在工作中经常会碰到需要进行异步消息处理的业务场景，根据消息性质的不同有完全不同的处理方式。

1、消息不独立

不独立的消息通常是有顺序依赖关系，这时消息处理机制将退化为线性队列处理模式，只能由一个消费者去单线程处理消息。

2、消息完全独立

完全独立的消息，可以由多个消费者（线程）并发同时处理，可以达到最大的并发处理能力。

3、消息不完全独立

通常这种情况是，同源消息（来自同一生产者）要求有序，异源消息顺序无关。

这个场景的消息处理会相对复杂点，为了保证同源消息有序，很容易想到对同一来源的消息绑定固定的消费者线程，这样做很简单但存在很大问题。

如果生产者数量很大，绑定线程数可能不够，当然可以复用线程资源，同一线程绑定多个消息来源进行处理，这样做又会有另一个问题：消息源之间的相互影响。

考虑以下场景：

生产者P1产生大量消息进入队列后被分配给消费线程C1处理（C1可能需要处理很长时间），这时生产者P2产生了一个消息，不幸的是也被分配给了消费线程C1处理

那么生产者P2的消息处理将被P1的大量消息给阻塞住，导致了P1和P2之间的相互影响，而且也不能充分利用其它消费线程导致不均衡。

所以，我们必须考虑避免这样的问题。做到消费处理的及时性（尽快）、隔离性（避免相互干扰）、均衡性（最大化并发处理）

在实现中，会有两种模式，比较容易想到的是线程派发模型（PUSH方式），具体做法通常如下：

1. 有一个全局消息派发者，轮询队列取出消息。

2. 根据消息来源，派发给合适的消费线程处理。

派发的算法机制简单的可以类似像基于消息来源的Hash，复杂的可以根据各个消费线程的当前负载，等待队列长短、消息的复杂度进行综合分析选择派发。

简单Hash肯定会碰到上述场景描述的问题，但复杂派发计算很明显实现起来非常麻烦和复杂，效率也不一定好，在均衡性方面也很难做到十分平衡。

第二种模式采用PULL方式，线程按需拉取，具体做法如下：

1. 消息源直接将产生的消息放入对应该源的临时队列中（如下所示每个session代表一个不同的消息来源），再将session置入一个阻塞队列通知线程处理

2.  多个消费线程同时轮询队列，争抢消息（保证只有一个线程取到

3. 检查队列指示器是否正被其他线程处理（实现时需要在线程级别基于同源消息的检测同步）

4. 若未被其他线程处理，则在同步区置处理中指示状态，退出同步区后对临时队列中的消息进行处理

5. 处理完成后，最后再次进入同步区置处理指示状态为空闲

下面用一段代码来描述下消费线程处理流程：

public void run() {
	try {
		for (AbstractSession s = squeue.take(); s != null; s = squeue.take()) {					
			// first check any worker is processing this session? 
                        // if any other worker thread is processing this event with same session, just ignore it.
			synchronized (s) {
				if (!s.isEventProcessing()) {
					s.setEventProcessing(true);
				} else {
					continue;
				}
			}
					
			// fire events with same session
			fire(s);
					
			// last reset processing flag and quit current thread processing
			s.setEventProcessing(false);
					
			// if remaining events, so re-insert to session queue
			if (s.getEventQueue().size() > 0 && !s.isEventProcessing()) {
				squeue.offer(s);
			}
		}
	} catch (InterruptedException e) {
		LOG.warn(e.getMessage(), e);
	}
}

关于“Springboot异步消息处理的方法”的内容就介绍到这里了，感谢大家的阅读。如果想了解更多行业相关的知识，可以关注创新互联行业资讯频道，小编每天都会为大家更新不同的知识点。