5

我正在尝试使用带有 Spring Data Redis 的 Redis 发布/订阅来实现聊天。

我使用 RedisTemplate 发布消息，如下所示：

public class RedisPublisher {

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    public void publish(ChannelTopic channelTopic, Object channelMessage) {
        redisTemplate.convertAndSend(channelTopic.getTopic(), channelMessage);
    }
}

为了接收消息，我有一个 MessageListener，如下所示：

public class RedisConsumer implements MessageListener {

MessageSerializer serializer = new MessageSerializer();
AtomicInteger atomicInteger = new AtomicInteger(0);

@Override
public void onMessage(Message message, byte[] pattern) {

    Object obj = serializer.deserialize(message.getBody());
    if(obj != null && obj instanceof RedisMessage) {
        System.err.println("Received message(" + atomicInteger.incrementAndGet() + ") " + obj.toString());
    }

}

消息是这样发布的：

final ChannelTopic channelTopic=connectionManager.subscribe("topic");
    new Thread(new Runnable() {
        public void run() {                
            Thread.sleep(5000);                
            for (int i = 0; i < 10; i++) {
                redisPublisher.publish(channelTopic, new RedisMessage(i + 1));
            }
        }
    }).run();

但是，收到的消息似乎以错误的顺序传递：

Received message(1) message id: 3
Received message(2) message id: 2
Received message(3) message id: 1
Received message(4) message id: 4
Received message(5) message id: 5
Received message(6) message id: 6
Received message(7) message id: 7
Received message(8) message id: 8
Received message(9) message id: 9
Received message(10) message id: 10

是否可以使用 Spring 提供的 RedisTemplate/MessageListener 同步发送/接收消息？

当前的代码库很小，可以在GitHub 上查看。

4

1 回答 1

6

众所周知，Redis PubSub 会按顺序传递消息（至少在您使用一个连接并触发时保证PUBLISH。该PUBLISH命令返回被通知的客户端数量）。乱序的原因是 Spring Data Redis 默认调度消息的方式。通知是在不同的线程上处理的，这就是原因。感谢您的代码，它帮助我快速重现该行为。

我可以想到两种可能的策略来解决这个问题：

  1. 但是，您可以提供一个执行RedisMessageListenerContainer. 现在，我想到的任何形式的同步都会损害性能。

  2. 在BinaryJedisPubSub. 您可以控制消息，并且可以省略执行程序问题。

HTH，马克

于 2015-08-26T18:55:41.943 回答