springboot中使用rabbitt的詳細(xì)方法

更新時(shí)間：2023年05月25日 15:09:40 作者：我想要她

這篇文章主要介紹了springboot中使用rabbitt,通過(guò)本文學(xué)習(xí)讓我們了解如何在Spring Boot中使用RabbitMQ，并使用不同的交換機(jī)和隊(duì)列類型以及消息確認(rèn)模式，需要的朋友可以參考下

RabbitMQ的示例，涉及到Direct、Fanout、Topic和Headers交換機(jī)以及普通隊(duì)列、延遲隊(duì)列和死信隊(duì)列

在pom.xml文件中添加以下依賴：

 <dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

配置RabbitMQ連接信息,在application.properties文件中添加以下配置：

spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

創(chuàng)建消息隊(duì)列

創(chuàng)建一個(gè)普通的Direct交換機(jī)和隊(duì)列：

 @Configuration
public class RabbitMQConfig {
    @Bean
    public Queue queue() {
        return new Queue("directQueue", false);
    }
    @Bean
    public DirectExchange directExchange() {
        return new DirectExchange("directExchange");
    }
    @Bean
    public Binding binding(Queue queue, DirectExchange directExchange) {
        return BindingBuilder.bind(queue).to(directExchange).with("directRoutingKey");
    }
}

創(chuàng)建一個(gè)Fanout交換機(jī)和隊(duì)列：

@Configuration
public class RabbitMQConfig {
    @Bean
    public Queue queueA() {
        return new Queue("fanoutQueueA", false);
    }
    @Bean
    public Queue queueB() {
        return new Queue("fanoutQueueB", false);
    }
    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange("fanoutExchange");
    }
    @Bean
    public Binding bindingA(Queue queueA, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(queueA).to(fanoutExchange);
    }
    @Bean
    public Binding bindingB(Queue queueB, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(queueB).to(fanoutExchange);
    }
}

創(chuàng)建一個(gè)Topic交換機(jī)和隊(duì)列：

@Configuration
public class RabbitMQConfig {
    @Bean
    public Queue queueA() {
        return new Queue("topicQueueA", false);
    }
    @Bean
    public Queue queueB() {
        return new Queue("topicQueueB", false);
    }
    @Bean
    public TopicExchange topicExchange() {
        return new TopicExchange("topicExchange");
    }
    @Bean
    public Binding bindingA(Queue queueA, TopicExchange topicExchange) {
        return BindingBuilder.bind(queueA).to(topicExchange).with("topic.key.*");
    }
    @Bean
    public Binding bindingB(Queue queueB, TopicExchange topicExchange) {
        return BindingBuilder.bind(queueB).to(topicExchange).with("topic.#");
    }
}

創(chuàng)建一個(gè)Headers交換機(jī)和隊(duì)列：

@Configuration
public class RabbitMQConfig {
    @Bean
    public Queue queue() {
        return new Queue("headerQueue", false);
    }
    @Bean
    public HeadersExchange headersExchange() {
        return new HeadersExchange("headerExchange");
    }
    @Bean
    public Binding binding(Queue queue, HeadersExchange headersExchange) {
        Map<String, Object> headers = new HashMap<>();
        headers.put("header1", "value1");
        headers.put("header2", "value2");
        return BindingBuilder.bind(queue).to(headersExchange).whereAll(headers).match();
    }
}

創(chuàng)建一個(gè)延遲隊(duì)列和死信隊(duì)列：

@Configuration
public class RabbitMQConfig {
    @Bean
    public Queue delayQueue() {
        Map<String, Object> args = new HashMap<>();
        args.put("x-message-ttl", 10000); // 消息過(guò)期時(shí)間為10秒
        args.put("x-dead-letter-exchange", "deadLetterExchange");
        args.put("x-dead-letter-routing-key", "deadLetterRoutingKey");
        return new Queue("delayQueue", false, false, false, args);
    }
    @Bean
    public Queue deadLetterQueue() {
        return new Queue("deadLetterQueue", false);
    }
    @Bean
    public DirectExchange directExchange() {
        return new DirectExchange("directExchange");
    }
    @Bean
    public DirectExchange deadLetterExchange() {
        return new DirectExchange("deadLetterExchange");
    }
    @Bean
    public Binding binding(Queue delayQueue, DirectExchange directExchange) {
        return BindingBuilder.bind(delayQueue).to(directExchange).with("delayRoutingKey");
    }
    @Bean
    public Binding deadLetterBinding(Queue deadLetterQueue, DirectExchange deadLetterExchange) {
        return BindingBuilder.bind(deadLetterQueue).to(deadLetterExchange).with("deadLetterRoutingKey");
    }
}

發(fā)送和接收消息

@Service
public class RabbitMQService {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    public void sendMessage(String message) {
        rabbitTemplate.convertAndSend("directExchange", "directRoutingKey", message);
    }
    @RabbitListener(queues = "directQueue")
    public void receiveDirectMessage(String message) {
        System.out.println("Received message from directQueue: " + message);
    }
    public void sendFanoutMessage(String message) {
        rabbitTemplate.convertAndSend("fanoutExchange", "", message);
    }
    @RabbitListener(queues = "fanoutQueueA")
    public void receiveFanoutMessageA(String message) {
        System.out.println("Received message from fanoutQueueA: " + message);
    }
    @RabbitListener(queues = "fanoutQueueB")
    public void receiveFanoutMessageB(String message) {
        System.out.println("Received message from fanoutQueueB: " + message);
    }
    public void sendTopicMessage(String message) {
        rabbitTemplate.convertAndSend("topicExchange", "topic.key.message", message);
    }
    @RabbitListener(queues = "topicQueueA")
    public void receiveTopicMessageA(String message) {
        System.out.println("Received message from topicQueueA: " + message);
    }
    @RabbitListener(queues = "topicQueueB")
    public void receiveTopicMessageB(String message) {
        System.out.println("Received message from topicQueueB: " + message);
    }
    public void sendHeaderMessage(String message) {
        MessageProperties messageProperties = new MessageProperties();
        messageProperties.setHeader("header1", "value1");
        messageProperties.setHeader("header2", "value2");
        Message msg = MessageBuilder.withBody(message.getBytes()).andProperties(messageProperties).build();
        rabbitTemplate.send("headerExchange", "", msg);
    }
    @RabbitListener(queues = "headerQueue")
    public void receiveHeaderMessage(Message message) {
        System.out.println("Received message from headerQueue: " + new String(message.getBody()));
    }
    public void sendDelayMessage(String message) {
        rabbitTemplate.convertAndSend("directExchange", "delayRoutingKey", message);
    }
    @RabbitListener(queues = "deadLetterQueue")
    public void receiveDeadLetterMessage(String message) {
        System.out.println("Received message from deadLetterQueue: " + message);
    }
}

消息確認(rèn)模式

在RabbitMQ中，消息確認(rèn)模式有兩種：自動(dòng)確認(rèn)和手動(dòng)確認(rèn)。自動(dòng)確認(rèn)是指當(dāng)消息被成功接收時(shí)，RabbitMQ會(huì)自動(dòng)確認(rèn)消息。手動(dòng)確認(rèn)是指當(dāng)消費(fèi)者成功處理消息后，顯式地向RabbitMQ發(fā)送確認(rèn)消息。
1 使用手動(dòng)確認(rèn)模式：

@Configuration
	public class RabbitMQConfig {
    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        return factory;
    }
}

@Service
public class RabbitMQService {
    @RabbitListener(queues = "directQueue")
    public void receiveDirectMessage(Message message, Channel channel) throws IOException {
        try {
            System.out.println("Received message from directQueue: " + new String(message.getBody()));
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
        }
    }
}

在上述代碼中，我們使用了MANUAL（手動(dòng)確認(rèn)）模式，并在消息處理成功后使用channel.basicAck()方法顯式地確認(rèn)消息。如果發(fā)生任何異常，我們使用channel.basicNack()方法拒絕消息并重新加入隊(duì)列。

希望這個(gè)示例能夠幫助您了解如何在Spring Boot中使用RabbitMQ，并使用不同的交換機(jī)和隊(duì)列類型以及消息確認(rèn)模式。

到此這篇關(guān)于springboot中使用rabbitt的文章就介紹到這了,更多相關(guān)springboot使用rabbitt內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: