java - Spring XD 中的错误处理

Question

我已将 Redis 配置为我的 spring-xd 设置的 MessageBus。当我的流失败时，数据被推送到错误队列。我正在尝试将它们读回并将它们推回目标队列。但是我没有看到我的 Sink 模块接收数据。有人可以帮助我了解哪里出了问题吗。

代码片段。

public RedisTemplate<String, byte[]> redisTemplate(RedisConnectionFactory redisConnectionFactory) {
  final RedisTemplate<String, byte[]> template = new RedisTemplate<String, byte[]>();
  template.setConnectionFactory(redisConnectionFactory);
  template.setKeySerializer(new StringRedisSerializer());
  template.setEnableDefaultSerializer(false);
  return template;
}

List<String> listOfKeys = new ArrayList<>();
Set<byte[]> keys = redisTemplate.getConnectionFactory().getConnection().keys("ERRORS*".getBytes());
for (byte[] data : keys) {
  listOfKeys.add(new String(data, 0, data.length));
}
for (String errorQueue : listOfKeys) {
    String destinationQueue = errorQueue.replace("ERRORS:", EMPTY_STRING);
    Long size = redisTemplate.opsForList().size(errorQueue);
    for (int i = 0; i < size; i++) {
        byte[] errorEvt = redisTemplate.opsForList().rightPop(errorQueue);
        redisTemplate.opsForList().leftPush(destinationQueue, errorEvt);
    }
}

Answer 1

快速浏览一下您的代码表明它应该可以正常工作；我建议你运行 redis-cli 然后输入 monitor 来观察redis流量。

编辑

要在下面回答您的评论 - 这取决于您的内容类型。如果是简单的文本，相对容易:

"\xff\x01\x0bcontentType\x00\x00\x00\x0c\"text/plain\"2016-05-20 10:11:07"

第一部分是标题，您可以使用 XD 的 EmbeddedHeadersMessageConverter 对其进行解码。

如果您的有效负载是一个 java 对象，它会被总线使用 kryo 序列化。