设置Spring Cloud RabbitMQ死信

Question

我有RabbitMQ的工作配置-使用TTL从等待（死信）的主队列中发送消息，并在到期时将其抛出：

@Configuration
@AllArgsConstructor
public class RabbitConfig {

public static final String QUEUE_MESSAGES = "demo-messages-queue";
public static final String QUEUE_MESSAGES_DLQ = QUEUE_MESSAGES + ".dlq";
public static final String EXCHANGE_MESSAGES = "demo-messages-exchange";

@Bean
public DirectExchange exchange() {
    return new DirectExchange(EXCHANGE_MESSAGES);
}

@Bean
public Queue mainQueue() {
    return QueueBuilder
            .durable(QUEUE_MESSAGES)
            .withArgument("x-dead-letter-exchange", EXCHANGE_MESSAGES)
            .withArgument("x-dead-letter-routing-key", QUEUE_MESSAGES_DLQ)
            .build();
}

@Bean
public Queue deadLetterQueue() {
    return QueueBuilder
            .durable(QUEUE_MESSAGES_DLQ)
            .withArgument("x-dead-letter-exchange", EXCHANGE_MESSAGES)
            .withArgument("x-dead-letter-routing-key", QUEUE_MESSAGES)
            .withArgument("x-message-ttl", 30000)
            .build();
}

@Bean
public Binding mainBinding() {
    return BindingBuilder
            .bind(mainQueue())
            .to(exchange())
            .with(QUEUE_MESSAGES);
}

@Bean
public Binding deadLetterBinding() {
    return BindingBuilder
            .bind(deadLetterQueue())
            .to(exchange())
            .with(QUEUE_MESSAGES_DLQ);
 }
}

对于应用程序中的Spring云流，我不能重复相同的方案。yml公司

spring:
cloud:
stream:
  binders:
    rabbitAll:
      type: rabbit
      environment:
        spring:
          rabbitmq:
            host: localhost
            port: 5672
            virtual-host: test
            username: guest
            password: guest
            message-ttl: 90000
  rabbit:
    bindings:
      demo-push-output:
        exchange-type: direct
        destination: demo-message-queue
        producer:
          binding-routing-key: 'demo-message-queue'
          routing-key-expression: '''demo-message-queue'''
      demo-push-input:
        consumer:
          bindingRoutingKey: demo-message-queue
          routingKeyExpression: '''demo-message-queue'''
          deadLetterRoutingKey: demo-message-queue.dlq
          deadLetterQueueName: demo-message-queue.dlq
          deadLetterExchange: demo-message-exchange
          dlqDeadLetterExchange: demo-message-exchange
          dlqDeadLetterRoutingKey: demo-message-queue
          autoBindDlq: true
          dlqTtl: 10000
  bindings:
    demo-push-input:
      destination: demo-message-queue
      group: demo-message-queueGroup
      binder: rabbitAll
    demo-push-output:
      destination: demo-message-queue
      group: demo-message-queueGroup
      binder: rabbitAll

我将非常感谢您在应用程序中提供的帮助和工作示例。yml公司

Answer 1

工作配置Spring-Cloud-Stream：

spring:
  cloud:
    stream:
      binders:
        rabbitAll:
          type: rabbit
          environment:
            spring:
              rabbitmq:
                host: localhost
                port: 5672
                virtual-host: test
                username: guest
                password: guest
                message-ttl: 90000
      rabbit:
        bindings:
          demo-push-output:
            exchangeType: direct
            destination: demo-message-queue
            producer:
              bindingRoutingKey: 'demo-message-queue'
              routingKeyExpression: '''demo-message-queue'''
          demo-push-input:
            consumer:
              bindingRoutingKey: demo-message-queue
              routingKeyExpression: '''demo-message-queue'''
              deadLetterRoutingKey: demo-message-queue.dlq
              deadLetterQueueName: demo-message-queue.dlq
              deadLetterExchange: demo-message-queue
              dlqDeadLetterExchange: demo-message-exchange
              dlqDeadLetterRoutingKey: demo-message-queue
              autoBindDlq: true
              dlqTtl: 30000
      bindings:
        demo-push-input:
          destination: demo-message-exchange
          group: demo-message-exchangeGroup
          binder: rabbitAll
          consumer:
            max-attempts: 1
        demo-push-output:
          destination: demo-message-exchange
          group: demo-message-exchangeGroup
          binder: rabbitAll

界面：

public interface DemoPushEventsSource {

String PUSH_INPUT = "demo-push-input";
String PUSH_OUTPUT = "demo-push-output";

@Input(PUSH_INPUT)
SubscribableChannel demoPushInput();

@Output(PUSH_OUTPUT)
MessageChannel demoPushOutput();

}

绑定：

@EnableBinding({DemoPushEventsSource.class})
public class StreamBindingConfiguration {
}

生产商：

public void sendMessage(User user) {
   source.demoPushOutput().send(buildMessageWithHeaders(user, headers));
}

public Message buildMessageWithHeaders(Object request, Map<String, Integer> headers) {
    return MessageBuilder.withPayload(request).copyHeaders(headers).build();
}

消费者：

@StreamListener(DemoPushEventsSource.PUSH_INPUT)
public void process(@Payload User model, @Headers Map<String, String> headers) throws Exception {
    throw new Exception("error exception");
}

Answer 2

请参阅文档。

具体来说dlqDeadLetterExchange和dlqDeadLetterRoutingKey。

您已经获得了dlqTtl。