redis订阅服务器无法与redis publisher一起使用

Question

我现在正在使用java设计redis pub/sub系统，但遇到了一个问题。我将向您展示细节：

出版商：

public class RedisMessagePublisher implements MessagePublisher {

public RedisMessagePublisher(StringRedisTemplate redisTemplate,ChannelTopic topic)
{
    this.redisTemplate = redisTemplate;
    this.topic = topic;
}

private StringRedisTemplate redisTemplate;

private ChannelTopic topic;

@Override
public void publish(String message) {
    redisTemplate.convertAndSend(topic.getTopic(), message);
    }
}

发布者是正确的，可以正确工作。

然后让我们转到subscriber类：

public class RedisMessageSubscriber implements MessageListener {

//action inspect here
private Action2<Message, byte[]> action;

public void setAction(Action2<Message, byte[]> action) {
    logger.info("action set");
    this.action = action;
}

private static Logger logger = LogManager.getLogger(RedisMessageSubscriber.class);

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

    logger.info("===> redis subscribe message in <===");

    if (action != null)
        action.call(message, bytes);
    else
        logger.info("===> action is null <===");
    }
}

在订阅类中，我使用RxJava注入Action，这样我可以更容易地使用它。

但问题是，在我发布了来自publisher的消息后，我可以确认消息可以传输到onMessage方法，日志打印不是我所期望的：

===> redis subscribe message in <===
===> action is null <===

我所期望的是，当我发布一条新消息时，订阅者得到了它并运行了我创建的操作。

我用来触发下面的发布者和订阅者的服务：

@RestController("redispubsubcontroller")
@RequestMapping(value = "/redis")
public class redispubsubcontroller {

@Autowired
private RedisMessagePublisher redisMessagePublisher;

@Autowired
private RedisMessageSubscriber redisMessageSubscriber;

private static Logger logger = LogManager.getLogger(redispubsubcontroller.class);

@RequestMapping(value = "/publisher", method = {RequestMethod.GET})
public ApiResponse getConfig(String message,HttpServletRequest request,
                                            HttpServletResponse response) {

    redisMessageSubscriber.setAction(new Action2<Message, byte[]>() {
        @Override
        public void call(Message message, byte[] bytes) {
            ObjectMapper objectMapper = new ObjectMapper();
            try {
                String result = objectMapper.readValue(message.getBody(), String.class);
                logger.info("receive："+result);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    });

    redisMessagePublisher.publish(message);

    return new ApiResponse("success","message sent");
    }
}

从上面的代码，你可以c我订阅了主题，并为订阅者设置了一个新操作：

 redisMessageSubscriber.setAction(new Action2<Message, byte[]>() {
    @Override
    public void call(Message message, byte[] bytes) {
        ObjectMapper objectMapper = new ObjectMapper();
        try {
            String result = objectMapper.readValue(message.getBody(), String.class);
            logger.info("receive："+result);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
});

但是我不知道为什么，在触发发布者之后，订阅者可以得到消息，但是仍然保持NULL Action，我创建的Action没有传递给它。

有人能帮忙吗？这个机制有问题吗？

====编辑=====

RedisMessageConfig代码如下：

@Configuration
public class RedisMessageConfig {

@Bean
ChannelTopic topic() {
    return new ChannelTopic("useraddresspubsub:queue");
}

@Bean
MessageListenerAdapter messageListener() {
    return new MessageListenerAdapter(new RedisMessageSubscriber());
}

@Autowired
private RedisConnectionFactory JedisConnectionFactory;

@Bean
RedisMessageListenerContainer redisContainer() {
    final RedisMessageListenerContainer container = new RedisMessageListenerContainer();
    container.setConnectionFactory(JedisConnectionFactory);
    container.addMessageListener(messageListener(), topic());
    return container;
    }
}

====解决====

最后我根据mp的想法解决了这个问题，将myredismessagesubscriber稍微更改为myredismessageconfig，因为流程是从redismessageconfig到redismessagesubscriber的，所以在redismessageconfig中，我需要首先向其注入操作，然后redismessageconfig将创建新的redismessagesubscriber并保留新创建的操作。代码如下：

@Component
public class MyRedisMessageConfig extends RedisMessageConfig {

private static Logger logger =LogManager.getLogger(MyRedisMessageConfig.class);

public MyRedisMessageConfig() {
    super.action = new Action2<Message, byte[]>() {
        @Override
        public void call(Message message, byte[] bytes) {
            String result = new String(message.getBody());
                logger.info("received：" + result);
            }
        };
    }
}

Answer 1

这不是MessageListener的工作方式。此外，还可以创建共享的可变状态。两次并发调用同时更改RedisMessageSubscriber的状态。

我假设在一个线程中设置操作，而在另一个线程上接收消息时，会遇到可见性问题。

如果每个MessageListener需要不同的行为，则创建实现该行为的多个侦听器。