本系列是学习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 fort_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:5672username: guestpassword: guestvirtual-host: /##开启Publisher Confirm机制publisher-confirms: true##开启Publisher Return机制publisher-returns: truetemplate:mandatory:truedatasource:url: jdbc:mysql://localhost:3306/rabbitmq?useUnicode=true&characterEncoding=utf-8&zeroDateTimeBehavior=convertToNull
username: rootpassword: binzhangdriverClassName: com.mysql.jdbc.Drivertype: com.alibaba.druid.pool.DruidDataSource
server:port:8001servlet:context-path: /mybatis:mapper-locations: classpath:mapping/*.xmllogging:level:tk:mybatis: trace

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

importcom.imooc.mq.entity.Order;importorg.springframework.amqp.rabbit.connection.CorrelationData;importorg.springframework.amqp.rabbit.core.RabbitTemplate;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.stereotype.Component;/*** @Title: OrderSender* @Description: 订单发送* @date 2019/1/2210:20*/@Componentpublic classOrderSender {//使用rabbitmq模板
@AutowiredprivateRabbitTemplate rabbitTemplate;//发送消息public void sendOrder(Order order) throwsException{CorrelationData correlationData= newCorrelationData();correlationData.setId(order.getMessageId());rabbitTemplate.convertAndSend("order-exchange",//exchange"order.abcd",//routingKeyorder,//消息体内容correlationData); //消息唯一id
}
}

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

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

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

@RunWith(SpringRunner.class)
@SpringBootTestpublic classMqApplicationTests {@AutowiredprivateOrderSender orderSender;@Testpublic voidcontextLoads() {Order order= newOrder();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:5672username: guestpassword: guest##连接到RabbitMQ的虚拟主机virtual-host: /## 消费端配置listener:simple:##消费者的最小数量concurrency:5## manual:手动 ack（确认）acknowledge-mode: manual##消费者的最大数量max-concurrency: 10##在单个请求中处理的消息个数，应该大于等于事务数量prefetch:1datasource:url: jdbc:mysql://localhost:3306/rabbitmq?useUnicode=true&characterEncoding=utf-8&zeroDateTimeBehavior=convertToNull
username: rootpassword: binzhangdriverClassName: com.mysql.jdbc.Driver
server:port:8002servlet:context-path: /

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

　　5、编写消息接收类

importcom.imooc.mq.entity.Order;importcom.rabbitmq.client.Channel;import org.springframework.amqp.rabbit.annotation.*;importorg.springframework.amqp.support.AmqpHeaders;importorg.springframework.messaging.handler.annotation.Headers;importorg.springframework.messaging.handler.annotation.Payload;importorg.springframework.stereotype.Component;importjava.util.Map;/*** @Title: OrderReceiver* @Description: 消费* @date 2019/1/2211:03*/@Componentpublic classOrderReceiver {/*** @RabbitListener 消息监听，可配置交换机、队列、路由key* 该注解会创建队列和交互机 并建立绑定关系* @RabbitHandler 标识此方法如果有消息过来，消费者要调用这个方法* @Payload 消息体* @Headers 消息头*@paramorder*/@RabbitListener(bindings=@QueueBinding(value= @Queue(value = "order-queue",declare = "true"),exchange= @Exchange(name = "order-exchange",declare = "true",type = "topic"),key= "order.abcd"))@RabbitHandlerpublic void onOrderMessage(@Payload Order order, @Headers Map<String,Object>headers,Channel channel)throwsException{//消费者操作System.out.println("------收到消息，开始消费------");System.out.println("订单ID："+order.getId());Long deliveryTag=(Long)headers.get(AmqpHeaders.DELIVERY_TAG);//现在是手动确认消息 ACKchannel.basicAck(deliveryTag,false);}
}

　　6、运行成功后

　

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