基于long pull实现简易的消息中心MQ参考

　　我们都用过消息中间件，它的作用自不必多说。但对于消费者却一直有一些权衡，就是使用push，还是pull模式的问题，这当然是各有优劣。当然，这并不是本文想讨论的问题。我们想在不使用长连接的情意下，如何实现实时的消息消费，而不至于让server端压力过大。大体上来说，这是一种主动拉取pull的方式。具体情况如何，且看且听。

 

1. 架构示意图

　　既然是一个消息中间的作用，我们必须得模拟一个生产消费者模型，如下：

 

 

 　　生产者集群->消息中心集群->消费者集群

　　只是这里的生产和消息中心也许我们可以合二为一，为简单起见，可能我们消费者只是想知道数据发生了变化。

　　以上是一个通用模型，接下来再说说我如何以long pull消息消费，其流程图如下：

 

 

 　　消费者一直请求连接->消息中心->有数据到来或者超时->消费者处理数据->发送ack确认->继续请求连接

　　如此一来，我们基本上就实现了一个消费模型了。但是有个问题，我们一直在不停地请求server，这会不会让server疲于奔命？是的，如果按照正常的http请求，就是不停地建立连接，处理数据，关闭连接等等。在没有消息到来之前，可以说，server会一直被这无用功跑死，它的qps越高，压力也越大。所以，我们使用了一种long pull的方式，让server端不要那么快返回没有意义的数据。但，这可能不是一件容易的事。

 

2. long pull的实现方式

　　long pull从原理上来说就是，必要的时候hold住连接，直到某个时机才返回。这和长链接有点类似。

　　至于为什么不用长连接实现，我想至少有两个原因：一是long pull一般基于http协议，实现简单且通用，而如果要基于长链接则需要了解太多的通信细节太复杂；二是端口复用问题，long pull可以直接基于业务端口实现，而长连接则必须要另外开一个通信端口，这在实际运维过程中也许不那么好操作，主要原因可能是我们往往不是真正的中间件，还达不到与架构或运维pk端口标准的资本。

　　说回正题，如何实现long pull？这其实和你使用的框架有关。但简单来说都可以这样干，请求进来后，我只要一直不返回即可。而且这也许是许多框架或语言的唯一选择。

　　如果咱们是java语言且基于spring系列框架，则可以用另外一种异步的方式。用上一种通用的实现方式的缺点是：当一个请求一直不返回后，必然占用主连接池，从而影响其他业务接口的请求处理，就是说只要你多接入几个这种请求，业务就别想有好日子过了。所以，我们选择异步的方式。异步，听起来是个好名词，但又该如何实现呢？我们普通异步，可能是直接丢到一个队列去，然后由后台线程一直处理即可，听起来不错。但这种请求至少两个问题：一是当我们提交到任务队列之后，连接还存在吗？二是我们敢让请求排队吗？因为如果排队有新数据进来，可就不面对实时的承诺了。

　　所以，针对上面的问题，spring系列有了解决方案。使用异步 servlet（async servlet），其操作步骤如下：

1 controller中返回异步实例callable；
2 在servlet中配置异步支持标识（统一配置）；

 

　　比如下面的demo：

// controller
    @GetMapping(value = "/consumeData")
    public Object consumeData(@RequestParam String topicName,
                              @RequestParam Long offset,
                              @RequestParam Long maxWait) {
        // 必要的时候需要在 web.xml中配置 <async-supported>true</async-supported>
        Callable<String> callable = () -> {
            SleepUtil.sleepMillis(10_000L);
            System.out.println("data come in, got out.");
            return "ok";
        };

        return callable;
    }
// web.xml
    // 所有需要的filter和servlet中，添加
    <async-supported>true</async-supported>

　　具体的框架版本各自具体配置可能不一样，自行查找资料即可。

　　以上，就解决了long pull的问题了。

 

3. 主键id的实现

　　主键id至少有两个作用：一是可用于唯一定位一条消息；二是可以用于去重做幂等；其实一般还有一个目的就是用于确认消息的先后顺序；

　　所以主键id很重要，往往需要经过精心的设计。但，我们这里可以简单的基于redis的自增key来处理即可。既保证了性能，又保证了唯一性，还保证了先后顺序问题。这就为后续消息的存储带来了方便。比如可以用zset存储这个消息id。

 

4. 数据到来的检测实现

　　在server端hold连接的同时，它又是如何发现数据已经到来了呢？

　　最简单的，可以让每个请求每隔一定时间，去查询一次数据，如果有则返回。但这个实现既不优雅也不经济也不实时，但是简单，可以适当考虑。

　　好点的方式，使用wait/notify机制，简单来说比如使用一个CountDownLatch，没有数据时则进行wait，数据到来时进行notify。这样下不来，不用每个请求反复查询数据，导致server压力变大，同时也让系统调度压力减小了，而且能够做到实时感知数据，可以说是很棒的选择。只是，这必然有很多的细节问题需要处理，稍有不慎，可能就是一个坑。比如：死锁问题，多节点问题，网络问题。。。 随便来一个，也许就jj了。

　　好好处理这个问题，总是好的。

 

5. 消息中心实现demo

 

5.1. 消费者生产者controller

　　两个简单方法入口，生产+消费 。

@RestController
@RequestMapping("/simpleMessageCenter")
public class SimpleMessageCenterController {

    @Resource
    private MessageService messageService;

    // 消费消息
    @GetMapping(value = "/consumeData")
    public Object consumeData(@RequestParam String topicName,
                              @RequestParam Long offset,
                              @RequestParam Long maxWait) {
        // 必要的时候需要在 web.xml中配置 <async-supported>true</async-supported>
        Callable<String> callable = () -> {
            try {
                Object data = messageService.consumeData(topicName, offset, maxWait);
                return JSONObject.toJSONString(data);
            }
            catch (Exception e){
                e.printStackTrace();
                return "error";
            }
        };

        return callable;
    }

    // 发送消息
    @GetMapping(value = "/sendMsg")
    public Object sendMsg(@RequestParam String topicName,
                          @RequestParam String extraId,
                          @RequestParam String data) {
        messageService.sendMsg(topicName, extraId, data);
        return "ok";
    }
}

 

5.2. 核心service简化版

　　由redis作为存储，展示各模块间的协作。

@Service
public class MessageService {

    @Resource
    private RedisTemplate<String, String> redisTemplate;

    // 消费闭锁
    private volatile ConcurrentHashMap<String, CountDownLatch>
            consumeLatchContainer = new ConcurrentHashMap<>();

    // 消费数据接口
    public List<Map<String, Object>> consumeData(String topic,
                                                 Long offset,
                                                 Long maxWait) throws InterruptedException {
        long startTime = System.currentTimeMillis();
        final CountDownLatch myLatch = getOrCreateConsumeLatch(topic);
        List<Map<String, Object>> result = new ArrayList<>();
        do {
            ZSetOperations<String, String> queueHolder
                    = redisTemplate.opsForZSet();
            Set<ZSetOperations.TypedTuple<String>> nextData
                    = queueHolder.rangeByScoreWithScores(topic, offset, offset + 100);
            if(nextData == null || nextData.isEmpty()) {
                long timeRemain = maxWait - (System.currentTimeMillis() - startTime);
                myLatch.await(timeRemain, TimeUnit.MILLISECONDS);
                continue;
            }
            for (ZSetOperations.TypedTuple<String> queue1 : nextData) {
                Map<String, Object> queueWrapped = new HashMap<>();
                queueWrapped.put(queue1.getValue(), queue1.getScore());
                result.add(queueWrapped);
            }
            break;
        } while (System.currentTimeMillis() - startTime <= maxWait);
        return result;
    }

    // 获取topic级别的锁
    private CountDownLatch getOrCreateConsumeLatch(String topicName) {
        return consumeLatchContainer.computeIfAbsent(
                    topicName, k -> new CountDownLatch(1));
    }

    // 接收到消息存储请求
    public void sendMsg(String topic, String extraIdSign, String data) {
        ValueOperations<String, String> strOp = redisTemplate.opsForValue();
        Long msgId = strOp.increment(topic + ".counter");
        // todo: 1. save real data
        // 2. 加入通知队列
        ZSetOperations<String, String> zsetOp = redisTemplate.opsForZSet();
        zsetOp.add(topic, extraIdSign, msgId);
        wakeupConsumers(topic, extraIdSign);
    }

    // 唤醒消费者，一般是有新数据到来
    private void wakeupConsumers(String topic, String extraIdSign) {
        CountDownLatch consumeLatch = getOrCreateConsumeLatch(topic);
        consumeLatch.countDown();
        rolloverConsumeLatch(topic, extraIdSign);
    }

    // 产生新一轮的锁
    private void rolloverConsumeLatch(String topic, String extraIdSign) {
        consumeLatchContainer.put(topic, new CountDownLatch(1));
    }
}

　　

5.3. 功能测试

　　因为是使用http接口实现，所以，可以直接通过浏览器实现功能测试。一个地址打开生产者链接，一个打开消费者链接。

// 1. 先访问消费者
http://localhost:8081/simpleMessageCenter/consumeData?topicName=q&offset=19&maxWait=50000
// 2. 再访问生产者
http://localhost:8081/simpleMessageCenter/sendMsg?topicName=q&extraId=d3&data=aaaaaaaaaaa

　　在生产者没有数据进来前，消费者会一直在等待，而生产者产生数据后，消费者就立即展示结果了。我们要实现的，不就是这个效果吗？

 

5.4. 消费者一直请求样例

　　在浏览器上我们看到的只是一次请求，但如果真正想实现，一直消费数据，则必须有一种订阅的感觉。其实就是不停的请求，处理，再请求的过程。

public class SimpleMessageCenterTest {

    @Test
    public void testConsumerSubscribe() {
        long offset = 0;
        String urlPrefix = "http://localhost:8081/simpleMessageCenter/consumeData?topicName=q&maxWait=50000&offset=";
        while (!Thread.interrupted()) {
            String dataListStr = HttpUtils.doGet(urlPrefix + offset);
            System.out.println("offsetStart: " + offset + ", got data:" + dataListStr);
            List<Object> dataListParsed = JSONObject.parseArray(dataListStr);
            // 不解析最终的offset了，大概就是根据最后一次offset再发起请求即可
            offset += dataListParsed.size();
        }
    }
}

　　以上，就是本次分享的小轮子了。我们抛却了消息系统中的一个重要且复杂的环节：存储。供参考。

不要害怕今日的苦，你要相信明天，更苦！