SpringBoot整合RabbitMQ-整合演示

2023-08-04 19:55:52

本系列是学习SpringBoot整合RabbitMQ的练手，包含服务安装，RabbitMQ整合SpringBoot2.x，消息可靠性投递实现等三篇博客。

　　学习路径：https://www.imooc.com/learn/1042 RabbitMQ消息中间件极速入门与实战

　　项目源码：https://github.com/ZbLeaning/Boot-RabbitMQ

整合实际上主要两步：

　　1、引入相关依赖

　　2、对application.yml进行配置

注意：后续需要使用数据库，因此需要安装mysql。https://blog.csdn.net/qq_37719778/article/details/81298292 mysql安装教程

Spring.RabbitMQ配置的含义可参考：https://blog.csdn.net/en_joker/article/details/80103519

数据库建表语句：

DROP TABLE IF EXISTS `broker_message_log`;

CREATE TABLE `broker_message_log` (

  `message_id` varchar(255) NOT NULL COMMENT '消息唯一ID',

  `message` varchar(4000) NOT NULL COMMENT '消息内容',

  `try_count` int(4) DEFAULT '0' COMMENT '重试次数',

  `status` varchar(10) DEFAULT '' COMMENT '消息投递状态 0投递中，1投递成功，2投递失败',

  `next_retry` timestamp NOT NULL DEFAULT '0000-00-00 00:00:00' ON UPDATE CURRENT_TIMESTAMP COMMENT '下一次重试时间',

  `create_time` timestamp NOT NULL DEFAULT '0000-00-00 00:00:00' ON UPDATE CURRENT_TIMESTAMP,

  `update_time` timestamp NOT NULL DEFAULT '0000-00-00 00:00:00' ON UPDATE CURRENT_TIMESTAMP,

  PRIMARY KEY (`message_id`)

) ENGINE=InnoDB DEFAULT CHARSET=utf8;

-- ----------------------------

-- Table structure for t_order

-- ----------------------------

DROP TABLE IF EXISTS `t_order`;

CREATE TABLE `t_order` (

  `id` int(11) NOT NULL AUTO_INCREMENT,

  `name` varchar(255) DEFAULT NULL,

  `message_id` varchar(255) DEFAULT NULL,

  PRIMARY KEY (`id`)

) ENGINE=InnoDB AUTO_INCREMENT=2018091102 DEFAULT CHARSET=utf8;

完成前期准备后开始进行整合。

Producer：服务端

　　1、新建一个SpringBoot项目，项目结构如下

　　2、添加Pom.xml文件依赖

 <!--rabbitmq依赖-->

        <dependency>

            <groupId>org.springframework.boot</groupId>

            <artifactId>spring-boot-starter-amqp</artifactId>

        </dependency>

        <dependency>

            <groupId>mysql</groupId>

            <artifactId>mysql-connector-java</artifactId>

        </dependency>

        <dependency>

            <groupId>org.mybatis.spring.boot</groupId>

            <artifactId>mybatis-boot-starter</artifactId>

        </dependency>

        <dependency>

            <groupId>tk.mybatis</groupId>

            <artifactId>mapper-spring-boot-starter</artifactId>

            <version>1.1.0</version>

        </dependency>

        <dependency>

            <groupId>com.alibaba</groupId>

            <artifactId>druid</artifactId>

            <version>1.0.29</version>

        </dependency>

        <dependency>

            <groupId>com.github.miemiedev</groupId>

            <artifactId>mybatis-paginator</artifactId>

            <version>1.2.17</version>

            <exclusions>

                <exclusion>

                    <groupId>org.mybatis</groupId>

                    <artifactId>mybatis</artifactId>

                </exclusion>

            </exclusions>

        </dependency>

        <!--工具类依赖包-->

        <dependency>

            <groupId>org.apache-commons</groupId>

            <artifactId>commons-lang3</artifactId>

        </dependency>

        <dependency>

            <groupId>commons-io</groupId>

            <artifactId>commons-io</artifactId>

            <version>2.4</version>

        </dependency>

        <dependency>

            <groupId>com.alibaba</groupId>

            <artifactId>fastjson</artifactId>

            <version>1.2.49</version>

        </dependency>

        <dependency>

            <groupId>javax.servlet</groupId>

            <artifactId>javax.servlet-api</artifactId>

            <scope>provided</scope>

        </dependency>

        <dependency>

            <groupId>log4j</groupId>

            <artifactId>log4j</artifactId>

            <version>1.2.17</version>

        </dependency>

　　3、配置application.yml文件

spring:

    rabbitmq:

        addresses: 134.175.33.221:5672

        username: guest

        password: guest

        virtual-host: /

        ##开启Publisher Confirm机制

        publisher-confirms: true

        ##开启Publisher Return机制

        publisher-returns: true

        template:

            mandatory: true

    datasource:

        url: jdbc:mysql://localhost:3306/rabbitmq?useUnicode=true&characterEncoding=utf-8&zeroDateTimeBehavior=convertToNull

        username: root

        password: binzhang

        driverClassName: com.mysql.jdbc.Driver

        type: com.alibaba.druid.pool.DruidDataSource

server:

    port: 8001

    servlet:

        context-path: /

mybatis:

    mapper-locations: classpath:mapping/*.xml

    logging:

        level:

            tk:

                mybatis: trace

　　4、编写消息发送类，直接使用SpringBoot配置的RabbitTemplate模板

import com.imooc.mq.entity.Order;

import org.springframework.amqp.rabbit.connection.CorrelationData;

import org.springframework.amqp.rabbit.core.RabbitTemplate;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.stereotype.Component;

/**

 * @Title: OrderSender

 * @Description: 订单发送

 * @date 2019/1/2210:20

 */

@Component

public class OrderSender {

    //使用rabbitmq模板

    @Autowired

    private RabbitTemplate rabbitTemplate;

　　　//发送消息

    public void sendOrder(Order order) throws Exception{

　　　　　

        CorrelationData correlationData = new CorrelationData();

        correlationData.setId(order.getMessageId());

        rabbitTemplate.convertAndSend("order-exchange",//exchange

                "order.abcd",//routingKey

                order,//消息体内容

                correlationData); //消息唯一id

    }

}

　　5、通过控制面板手动建立交换机exchange、消息队列queue

　　6、点击进入创建好的order-exchange，设置绑定路由键

　　7、写测试demo，运行消息发现，看控制台是否收到消息

@RunWith(SpringRunner.class)

@SpringBootTest

public class MqApplicationTests {

    @Autowired

    private OrderSender orderSender;

    @Test

    public void contextLoads() {

        Order order = new Order();

        order.setId("aaa");

        order.setName("测试消息a");

        order.setMessageId(System.currentTimeMillis() + "$" + UUID.randomUUID().toString());

        try {

            orderSender.sendOrder(order);

        } catch (Exception e) {

            e.printStackTrace();

        }

    }

}

　　接收消息成功：

　　注意：启动时如果报异常

Failed to configure a DataSource: 'url' attribute is not specified and no embedded datasource could be configured.

　　则需要配置

@MapperScan("com.imooc.mq.mapper") @SpringBootApplication(exclude = DataSourceAutoConfiguration.class)

Consumer：消费端

　　1、项目结构

　　2、配置Pom.xml文件，引入依赖，可直接复制服务端pom.xml注入的依赖

　　3、配置Application.xml文件

## springboot整合rabbitmq的基本配置

spring:

  rabbitmq:

    addresses: 134.175.33.221:5672

    username: guest

    password: guest

    ##连接到RabbitMQ的虚拟主机

    virtual-host: /

    ## 消费端配置

    listener:

      simple:

        ##消费者的最小数量

        concurrency: 5

        ## manual:手动 ack（确认）

        acknowledge-mode: manual

        ##消费者的最大数量

        max-concurrency: 10

        ##在单个请求中处理的消息个数，应该大于等于事务数量

        prefetch: 1

  datasource:

    url: jdbc:mysql://localhost:3306/rabbitmq?useUnicode=true&characterEncoding=utf-8&zeroDateTimeBehavior=convertToNull

    username: root

    password: binzhang

    driverClassName: com.mysql.jdbc.Driver

server:

  port: 8002

  servlet:

    context-path: /

　　4、需要将服务端的Order类也复制到消费端

　　5、编写消息接收类

import com.imooc.mq.entity.Order;

import com.rabbitmq.client.Channel;

import org.springframework.amqp.rabbit.annotation.*;

import org.springframework.amqp.support.AmqpHeaders;

import org.springframework.messaging.handler.annotation.Headers;

import org.springframework.messaging.handler.annotation.Payload;

import org.springframework.stereotype.Component;

import java.util.Map;

/**

 * @Title: OrderReceiver

 * @Description: 消费

 * @date 2019/1/2211:03

 */

@Component

public class OrderReceiver {

    /**

     * @RabbitListener 消息监听，可配置交换机、队列、路由key

     * 该注解会创建队列和交互机 并建立绑定关系

     * @RabbitHandler 标识此方法如果有消息过来，消费者要调用这个方法

     * @Payload 消息体

     * @Headers 消息头

     * @param order

     */

    @RabbitListener(bindings = @QueueBinding(

            value = @Queue(value = "order-queue",declare = "true"),

            exchange = @Exchange(name = "order-exchange",declare = "true",type = "topic"),

            key = "order.abcd"

    ))

    @RabbitHandler

    public void onOrderMessage(@Payload Order order, @Headers Map<String,Object> headers,

                               Channel channel) throws Exception{

        //消费者操作

        System.out.println("------收到消息，开始消费------");

        System.out.println("订单ID："+order.getId());

        Long deliveryTag = (Long)headers.get(AmqpHeaders.DELIVERY_TAG);

        //现在是手动确认消息 ACK

        channel.basicAck(deliveryTag,false);

    }

}

　　6、运行成功后

基本的服务和消费端整合及演示demo已完成，一般开发过程中我们大都采用手动确认消息机制，如果注释掉该行则会出现消息被消费但是一直处于未被确认的状态。当重启服务端再次发现消息时，消息也会被消费。

码农公寓

相关文章