Rocketmq 的 LitePullConsumer 使用

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81

@Slf4j
public class RocketMQHandler extends AbstractMQHandler {

  private final RocketMQConfig config;

  private final Supplier<DefaultMQProducer> producer;

  private final Supplier<DefaultLitePullConsumer> consumer;

  public RocketMQHandler(RocketMQProperties properties, RocketMQConfig config) {
    this.config = config;
    // 这里要延迟初始化，否则启动 consumer 占用 consumerQueue
    this.producer = SingletonSupplier.of(() -> createProducer(properties, config));
    this.consumer = SingletonSupplier.of(() -> createConsumer(properties, config));
  }

  @SneakyThrows
  @Override
  public void send(String message) {
    Message m = new Message(config.getDestinationName(), message.getBytes(StandardCharsets.UTF_8));
    if (log.isTraceEnabled()) {
      log.trace("{} send message: {}", RocketMQHandler.this.getClass().getSimpleName(), message);
    }
    producer.get().send(m);
  }

  @Override
  public void receive(MQListener listener) {
    ConsumerTask task = new ConsumerTask(listener);
    executorService.schedule(task, PULL_PERIOD, TimeUnit.MILLISECONDS);
  }

  @AllArgsConstructor
  private class ConsumerTask implements Runnable {

    private MQListener listener;

    @Override
    public void run() {
      List<MessageExt> msgs = consumer.get().poll(PULL_PERIOD);
      if (CollectionUtils.isNotEmpty(msgs)) {
        msgs.forEach(m -> {
          String message = new String(m.getBody(), StandardCharsets.UTF_8);
          if (log.isTraceEnabled()) {
            log.trace("{} receive message: {}", RocketMQHandler.this.getClass().getSimpleName(), message);
          }

          listener.onMessage(message);
        });
      }

      executorService.schedule(this, PULL_PERIOD, TimeUnit.MILLISECONDS);
    }
  }

  @SneakyThrows
  private DefaultMQProducer createProducer(RocketMQProperties properties, RocketMQConfig config) {
    DefaultMQProducer producer = new DefaultMQProducer(config.getProducerGroup());
    producer.setNamesrvAddr(properties.getNamesrvAddr());
    producer.start();
    return producer;
  }

  @SneakyThrows
  private DefaultLitePullConsumer createConsumer(RocketMQProperties properties, RocketMQConfig config) {
    DefaultLitePullConsumer consumer = new DefaultLitePullConsumer(config.getConsumerGroup());
    consumer.setNamesrvAddr(properties.getNamesrvAddr());
    switch (config.getConsumeMode()) {
      case P2P:
        consumer.setMessageModel(MessageModel.CLUSTERING);
        break;
      case BROADCAST:
        consumer.setMessageModel(MessageModel.BROADCASTING);
        break;
    }
    consumer.subscribe(config.getDestinationName(), "*");
    consumer.start();
    return consumer;
  }

}