rabbitmq自定义消息监听

喜欢ヅ旅行 2022-12-08 05:42 100阅读 0赞

package com.test.config;
    
    import com.test.adapter.MessageAdapter;
    import org.springframework.amqp.core.*;
    import org.springframework.amqp.rabbit.connection.ConnectionFactory;
    import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
    import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
    import org.springframework.beans.factory.annotation.Qualifier;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    @Configuration
    public class AmqpConfiguration {
    
        @Bean
        Exchange exchange(@Value("${config.mq.exchange}") String exchangeName) {
            return ExchangeBuilder.topicExchange(exchangeName).durable(true).build();
        }
    
        @Bean
        Queue realtimeQueue(@Value("${config.mq.realtime.queue}") String queueName) {
            return QueueBuilder.durable(queueName).build();
        }
    
        @Bean
        Binding realtimeBinding(
                Exchange exchange,
                @Qualifier("realtimeQueue") Queue queue,
                @Value("${config.mq.realtime.routingKey}") String routingKey,
                @Value("${config.mq.realtime.queue}") String queueName) {
            return BindingBuilder.bind(queue).to(exchange).with(routingKey).noargs();
        }
    
        @Bean
        SimpleMessageListenerContainer container(
            ConnectionFactory connectionFactory,
            MessageListenerAdapter listenerAdapter,
            @Value("${config.mq.realtime.queue}") String... queueName) {
            SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
            container.setConnectionFactory(connectionFactory);
            container.setQueueNames(queueName);
            container.setMessageListener(listenerAdapter);
            container.setConcurrentConsumers(3);
            return container;
        }
    
        @Bean
        MessageListenerAdapter messageListenerAdapter() {
            return new MessageListenerAdapter(messageAdapter(), "receive");
        }
    
        @Bean
        MessageAdapter messageAdapter() {
            return new MessageAdapter();
        }
    
    }

package com.test.adapter;
    
    import lombok.extern.slf4j.Slf4j;
    
    @Slf4j
    public class MessageAdapter {
    
        public void receive(String message) {
            log.info("receive message ==> {}", message);
        }
    
    }