`

SpringBoot集成RabbitMQ

 
阅读更多 
pom:
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
yml:
spring:
  rabbitmq:
    host: ip
    port: 5672
    virtual-host: /
    username: 账号
    password: 密码
 
1.直连型交换机
根据消息携带的路由键将消息投递给对应队列。
流程：有一个队列绑定到一个直连交换机上，同时设置一个路由键。然后当一个消息携带着路由值X，这个消息通过生产者发送给交换机，交换机就会根据这个路由值X去寻找绑定值也是X的队列。生产者：
//直连型交换机
@Configuration
public class DirectRabbitConfig {

    //队列 起名：TestDirectQueue
    @Bean
    public Queue TestDirectQueue() {
        // durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效。
        // exclusive:默认是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。
        // autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。//一般设置一下队列的持久化就好,其余两个就是默认false
        return new Queue("TestDirectQueue",true, false, false);
    }

    //Direct交换机 起名：TestDirectExchange
    @Bean
    DirectExchange TestDirectExchange() {
        return new DirectExchange("TestDirectExchange",true,false);
    }

    //绑定  将队列和交换机绑定, 并设置用于匹配键：TestDirectRouting
    @Bean
    Binding bindingDirect() {
        return BindingBuilder.bind(TestDirectQueue()).to(TestDirectExchange()).with("TestDirectRouting4");
    }

}
package com.demo.business.controller;

import com.demo.common.msg.BaseResponse;
import com.demo.common.util.ResponseMsgUtil;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

@RestController
@RequestMapping("/mq")
public class RabbitMQController {

    @Autowired
    RabbitTemplate rabbitTemplate;

    @GetMapping("/send")
    public BaseResponse send() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "test message, hello!";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String,Object> map=new HashMap<>();
        map.put("messageId",messageId);
        map.put("messageData",messageData);
        map.put("createTime",createTime);
        //将消息携带绑定键值：TestDirectRouting 发送到交换机TestDirectExchange
        rabbitTemplate.convertAndSend("TestDirectExchange", "TestDirectRouting", map);
        return ResponseMsgUtil.success();
    }
    
}

消费者：

package com.demo.business.controller;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.util.Map;

@Component
@RabbitListener(queues = "TestDirectQueue")//监听的队列名称 TestDirectQueue
public class DirectReceiver {

    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("DirectReceiver消费者收到消息  : " + testMessage.toString());
    }

}

 

2.主题交换机
主题交换机与直连交换机流程相似，区别在于绑定键有规则（*便是一个单词，#表示零个或多个）。生产者：
package com.demo.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class TopicRabbitConfig {
    //绑定键
    public final static String man = "topic.man";
    public final static String woman = "topic.woman";

    @Bean
    public Queue firstQueue() {
        return new Queue(TopicRabbitConfig.man);
    }

    @Bean
    public Queue secondQueue() {
        return new Queue(TopicRabbitConfig.woman);
    }

    @Bean
    TopicExchange exchange() {
        return new TopicExchange("topicExchange");
    }

    //将firstQueue和topicExchange绑定,而且绑定的键值为topic.man
    //这样只要是消息携带的路由键是topic.man,才会分发到该队列
    @Bean
    Binding bindingExchangeMessage() {
        return BindingBuilder.bind(firstQueue()).to(exchange()).with(man);
    }

    //将secondQueue和topicExchange绑定,而且绑定的键值为用上通配路由键规则topic.#
    // 这样只要是消息携带的路由键是以topic.开头,都会分发到该队列
    @Bean
    Binding bindingExchangeMessage2() {
        return BindingBuilder.bind(secondQueue()).to(exchange()).with("topic.#");
    }

}
@GetMapping("/send2")
    public BaseResponse send2() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "test message, hello!";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String,Object> map=new HashMap<>();
        map.put("messageId",messageId);
        map.put("messageData",messageData);
        map.put("createTime",createTime);
        //将消息携带绑定键值：TestDirectRouting 发送到交换机TestDirectExchange
        rabbitTemplate.convertAndSend("topicExchange", "topic.woman22222", map);
        return ResponseMsgUtil.success();
    }

消费者：

import java.util.Map;

@Component
@RabbitListener(queues = "topic.woman")//队列名称
public class TopicManReceiver {

    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("TopicManReceiver消费者收到消息  : " + testMessage.toString());
    }

}

 

3.扇形交换机
没有路由键的概念，接收到消息后，会直接转发到绑定到它上面的所有队列。生产者：
package com.demo.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class FanoutRabbitConfig {

    @Bean
    public Queue queueA() {
        return new Queue("fanout.A");
    }

    @Bean
    public Queue queueB() {
        return new Queue("fanout.B");
    }

    @Bean
    public Queue queueC() {
        return new Queue("fanoutC");
    }

    @Bean
    FanoutExchange fanoutExchange() {
        return new FanoutExchange("fanoutExchange");
    }

    @Bean
    Binding bindingExchangeA() {
        return BindingBuilder.bind(queueA()).to(fanoutExchange());
    }

    @Bean
    Binding bindingExchangeB() {
        return BindingBuilder.bind(queueB()).to(fanoutExchange());
    }

    @Bean
    Binding bindingExchangeC() {
        return BindingBuilder.bind(queueC()).to(fanoutExchange());
    }

}
@GetMapping("/send3")
public BaseResponse send3() {
　　String messageId = String.valueOf(UUID.randomUUID());
　　String messageData = "test message, hello!";
　　String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
　　Map<String,Object> map=new HashMap<>();
　　map.put("messageId",messageId);
　　map.put("messageData",messageData);
　　map.put("createTime",createTime);
　　//将消息携带绑定键值：TestDirectRouting 发送到交换机TestDirectExchange
　　rabbitTemplate.convertAndSend("fanoutExchange", null, map);
　　return ResponseMsgUtil.success();
}

消费者：

ackage com.demo.business.controller;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.util.Map;

@Component
@RabbitListener(queues = "fanout.A")
public class FanoutReceiverA {

    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("FanoutReceiverA消费者收到消息  : " +testMessage.toString());
    }

}
package com.demo.business.controller;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.util.Map;

@Component
@RabbitListener(queues = "fanoutC")
public class FanoutReceiverC {

    @RabbitHandler
    public void process(Map testMessage) {
        System.out.println("FanoutReceiverC消费者收到消息  : " +testMessage.toString());
    }

}

 另外，还有Header Exchange 头交换机、Default Exchange 默认交换机、Dead Letter Exchange 死信交换机。

4.消息回调
生产这发送消息结束后回调。yml增加配置：
spring:
    #确认消息已发送到交换机(Exchange)
    publisher-confirms: true
    #确认消息已发送到队列(Queue)
    publisher-returns: true
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
@Configuration
public class RabbitConfig {
 
    @Bean
    public RabbitTemplate createRabbitTemplate(ConnectionFactory connectionFactory){
        RabbitTemplate rabbitTemplate = new RabbitTemplate();
        rabbitTemplate.setConnectionFactory(connectionFactory);
        //设置开启Mandatory,才能触发回调函数,无论消息推送结果怎么样都强制调用回调函数
        rabbitTemplate.setMandatory(true);
 
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                System.out.println("ConfirmCallback:     "+"数据："+correlationData);
                System.out.println("ConfirmCallback:     "+"结果："+ack);
                System.out.println("ConfirmCallback:     "+"原因："+cause);
            }
        });
 
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                System.out.println("ReturnCallback:     "+"消息："+message);
                System.out.println("ReturnCallback:     "+"状态码："+replyCode);
                System.out.println("ReturnCallback:     "+"信息："+replyText);
                System.out.println("ReturnCallback:     "+"交换机："+exchange);
                System.out.println("ReturnCallback:     "+"键值："+routingKey);
            }
        });
 
        return rabbitTemplate;
    }
 
}

调用规则：

找不到交换机，调用：ConfirmCallback；
找到交换机，没找到队列,调用：ConfirmCallback和RetrunCallback；
交换机和队列都没找到，调用：ConfirmCallback；
消息推送成功，调用：ConfirmCallback；
5.消息确认
消费者收到消息后进行操作。
手动确认：消费者收到消息后手动调用basic.ack/basic.nack/basic.reject后，RabbitMQ收到这些消息后，才认为本次投递成功。
basicAck用于肯定确认
basicNack用于否定确认
basicReject用于否定确认，与basic.nack相比，一次只能拒绝单挑消息。
import com.elegant.rabbitmqconsumer.receiver.MyAckReceiver;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
@Configuration
public class MessageListenerConfig {
 
    @Autowired
    private CachingConnectionFactory connectionFactory;
    @Autowired
    private MyListenerConfig myListenerConfig;//消息接收处理类
 
    @Bean
    public SimpleMessageListenerContainer simpleMessageListenerContainer() {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        container.setConcurrentConsumers(1);
        container.setMaxConcurrentConsumers(1);
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);//RabbitMQ默认是自动确认，这里改为手动确认消息
        container.setQueueNames("TestDirectQueue");
        container.setMessageListener(myListenerConfig);
        return container;
    }
}

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.stereotype.Component;
import java.util.HashMap;
import java.util.Map;
 
@Component
public class MyListenerConfig implements ChannelAwareMessageListener {
 
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            System.out.println(消息："+message.toString());
            channel.basicAck(deliveryTag, true);
            //channel.basicReject(deliveryTag, true);//为true会重新放回队列
        } catch (Exception e) {
            channel.basicReject(deliveryTag, false);
            e.printStackTrace();
        }
    }

}

 

 

代码：https://files.cnblogs.com/files/DreamFather/mq.zip

分享到：
| protocol method: #method<channel.close>( ...
评论
发表评论

您还没有登录,请您登录后再发表评论

相关推荐

    SpringBoot集成RabbitMQ延时队列，自定义延时时间Demo

    该示例通过 rabbitmq_delayed_message_exchange 插件实现自定义延时时间的延时队列。 示例是纯净的，只引入了需要的架包 启动示例时，请确保MQ已经安装了延时插件(附件里带有插件及安装说明)以及示例的MQ相关的配置...

    Springboot整合RabbitMQ最简单demo

    Springboot整合RabbitMQ最简单demo,适用于springcloud项目,作为消息总线适用,需要安装RabbitMQ,Mac linux可以使用命令行一键安装,在项目配置文件配置好端口即可(已默认配置),启动项目访问8080端口,参数见controller.

    rabbitmq基础+springboot集成rabbitmq

    **SpringBoot集成RabbitMQ** Spring Boot提供了方便的RabbitMQ支持，简化了配置和操作。集成RabbitMQ需要以下步骤： 1. 添加依赖：在`pom.xml`中引入`spring-boot-starter-amqp`依赖。 2. 配置：在`application....

    SpringBoot集成RabbitMQ消息中间件.zip

    SpringBoot集成RabbitMQ消息中间件

    springboot集成rabbitmq的简单使用

    springboot集成rabbitmq的简单使用，介绍了springboot集成rabbitmq的使用，利用的交换机、队列、路由key来实现的例子

    SpringBoot整合RabbitMQ.zip

    SpringBoot整合RabbitMQ的详细过程 **1.该篇博文首先讲述了交换机和队列之间的绑定关系** ①direct、②fanout、③topic **2.然后讲消息的回调** 四种情况下，确认触发哪个回调函数： ①消息推送到server，但是在...

    springboot+rabbitmq项目demo(亲测,可正常运行)

    这个项目是一个基础的SpringBoot与RabbitMQ集成示例，展示了如何在Java环境中利用SpringBoot的便利性构建消息队列系统。实际应用中，你可能需要处理更复杂的情况，如多个消费者、交换机、路由键等。不过，这个简单的...

    springboot整合rabbitmq，开启手工确认。保证消息100%投递

    在Spring Boot应用中整合RabbitMQ，以确保消息100%投递，是一个关键的实践，特别是对于那些需要高可靠性和数据一致性的系统。RabbitMQ是一个流行的开源消息代理，它遵循Advanced Message Queuing Protocol (AMQP)...

    springboot集成rabbitMQ

    springboot集成rabbitmq消息队列示例

    Springboot+rabbitmq集成

    文件内包含了rabbit安装的必需文件以及springboot整合rabbitmq的完整代码，代码里包含了原生的rabbitmq使用代码和整合springboot后的使用代码，还有rabbit队列的所有消息队列模式，代码简单易懂，解压打开就可以使用

    基于SpringBoot整合RabbitMQ发送邮件通知

    本话题主要关注如何利用 SpringBoot 集成 RabbitMQ 来实现邮件通知功能，这在微服务间进行异步通信或触发业务流程时非常有用。 首先，让我们深入了解 SpringBoot。SpringBoot 是由 Pivotal 团队提供的全新框架，它...

    springboot整合rabbitMq和多数据源动态切换和跨域访问和gradle加maven和shiro安全框架和lombok自动getset生成

    标题中的“springboot整合rabbitMq和多数据源动态切换和跨域访问和gradle加maven和shiro安全框架和lombok自动getset生成”表明这是一个关于Spring Boot集成多个技术的项目。以下是对这些技术及其整合的详细解释： 1...

    SpringBoot集成Rabbitmq简单案例

    在压缩包`springBoot+rabbitmq`中，可能包含了上述代码的示例项目，你可以下载并运行它，以便更好地理解和实践这个简单的Spring Boot与RabbitMQ集成案例。通过这种方式，你可以快速上手并掌握Spring Boot中RabbitMQ...

    SpringBoot集成RabbitMQ代码示例

    本文将深入探讨如何在SpringBoot项目中集成RabbitMQ，通过代码示例来展示具体步骤和实现。 首先，集成RabbitMQ需要在SpringBoot的`pom.xml`文件中添加依赖。你需要引入`spring-boot-starter-amqp`依赖，这会包含...

    springboot+rabbitMQ+websocket

    首先，让我们了解Spring Boot如何集成RabbitMQ。在Spring Boot项目中引入`spring-boot-starter-amqp`依赖，它包含了对RabbitMQ的支持。接着，配置RabbitMQ的相关参数，如主机地址、端口、用户名和密码。在`...

    SpringBoot集成RabbitMQ实现用户注册的示例代码

    本文主要介绍了使用SpringBoot集成RabbitMQ实现用户注册的示例代码，通过详细的示例代码，帮助读者学习和工作。该示例代码介绍了如何使用RabbitMQ实现用户注册功能，提供了详细的实现步骤和代码示例。 SpringBoot...

    SpringBoot-RabbitMQ生产者和消费者.7z

    这个名为"SpringBoot-RabbitMQ生产者和消费者.7z"的压缩包包含两个关键部分：`rabbitmq-consumer`（消费者）和`rabbitmq-provider`（生产者），它们演示了如何在SpringBoot应用中集成并利用RabbitMQ进行数据交换。...

    springboot+rabbitmq实现延时队列

    本教程将详细介绍如何使用SpringBoot集成RabbitMQ来实现一个延时队列，并探讨消息发送与消费确认机制以及消费者端的策略模式应用。 首先，SpringBoot是Java开发者广泛使用的快速开发框架，它简化了Spring的配置和...

    springboot-rabbitmq.rar

    6. **SpringBoot集成RabbitMQ** 在SpringBoot中，我们可以通过添加`spring-boot-starter-amqp`依赖来引入RabbitMQ支持。然后，使用Spring的`@RabbitListener`和`RabbitTemplate`注解来定义消费者和生产者。此外，还...

Magicbox
Global site tag (gtag.js) - Google Analytics