本系列是學習SpringBoot整合RabbitMQ的練手，包含服務安裝，RabbitMQ整合SpringBoot2.x，消息可靠性投遞實現等三篇博客。

　　學習路徑：https://www.imooc.com/learn/1042?RabbitMQ消息中間件極速入門與實戰?

　　項目源碼：https://github.com/ZbLeaning/Boot-RabbitMQ?

---

?

設計一個消息可靠性投遞方案，服務結構如下：

?

?組成：

　　Sender+Confirm Listener ：組成消息的生產者

　　MQ Broker：消息的消費者，包含具體的MQ服務

　　BIZ DB：業務數據數據庫

　　MSG DB：消息日志記錄數據庫（0：發送中、1：發送成功、2：發送失敗）

思路：

　　以最常見的創建訂單業務來舉例，假設訂單創建成功后需要去發短信通知用戶

　　1、先完成訂單業務數據的存儲，并記錄這條操作日志（發送中）

　　2、生產者發送一條消息到消費者（異步）

　　3、消費者成功消費后給給Confirm listener發送應答

　　4、監聽收到消息確認成功后，對消息日志表操作，修改之前的日志狀態（發送成功）

　　5、在消費端返回應答的過程中，可能發生網絡異常導致生產者未收到應答消息，因此需要一個定時任務去撈取狀態是發送中并已經超時的消息集合

　　6、將撈取到的日志對應的消息，進行重發

　　7、定時任務判斷設置的消息最大重投次數，大于最大重投次數就判斷消息發送失敗，更新日志記錄狀態（發送失敗）

---

?項目搭建

　　Durid數據源配置文件

//druid.properties ##下面為連接池的補充設置，應用到上面所有數據源中 #初始化大小，最小，最大 druid.initialSize=5 druid.minIdle=10 druid.maxActive=300 #配置獲取連接等待超時的時間 druid.maxWait=60000 #配置間隔多久才進行一次檢測，檢測需要關閉的空閑連接，單位是毫秒 druid.timeBetweenEvictionRunsMillis=60000 #配置一個連接在池中最小生存的時間，單位是毫秒 druid.minEvictableIdleTimeMillis=300000 druid.validationQuery=SELECT 1 FROM DUAL druid.testWhileIdle=true druid.testOnBorrow=false druid.testOnReturn=false #打開PSCache，并且指定每個連接上PSCache的大小 druid.poolPreparedStatements=true druid.maxPoolPreparedStatementPerConnectionSize=20 #配置監控統計攔截的filters，去掉后監控界面sql無法統計，'wall'用于防火墻 druid.filters=stat,wall,log4j #通過connectProperties屬性來打開mergeSql功能；慢SQL記錄 druid.connectionProperties=druid.stat.mergeSql=true;druid.stat.slowSqlMillis=5000 #合并多個DruidDataSource的監控數據 druid.useGlobalDataSourceStat=true

　　添加相應的數據源配置類、定時任務配置類、常量類

package com.imooc.mq.config.database;import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.PropertySource; import org.springframework.context.support.PropertySourcesPlaceholderConfigurer; import org.springframework.stereotype.Component;/*** @Title: DruidDataSourceSettings* @Description: Druid數據源讀取配置* @date 2019/1/2214:31*/ @Component @ConfigurationProperties(prefix = "spring.datasource") @PropertySource("classpath:druid.properties") public class DruidDataSourceSettings {private String driverClassName;private String url;private String username;private String password;@Value("${druid.initialSize}")private int initialSize;@Value("${druid.minIdle}")private int minIdle;@Value("${druid.maxActive}")private int maxActive;@Value("${druid.timeBetweenEvictionRunsMillis}")private long timeBetweenEvictionRunsMillis;@Value("${druid.minEvictableIdleTimeMillis}")private long minEvictableIdleTimeMillis;@Value("${druid.validationQuery}")private String validationQuery;@Value("${druid.testWhileIdle}")private boolean testWhileIdle;@Value("${druid.testOnBorrow}")private boolean testOnBorrow;@Value("${druid.testOnReturn}")private boolean testOnReturn;@Value("${druid.poolPreparedStatements}")private boolean poolPreparedStatements;@Value("${druid.maxPoolPreparedStatementPerConnectionSize}")private int maxPoolPreparedStatementPerConnectionSize;@Value("${druid.filters}")private String filters;@Value("${druid.connectionProperties}")private String connectionProperties;@Beanpublic static PropertySourcesPlaceholderConfigurer properdtyConfigure(){return new PropertySourcesPlaceholderConfigurer();}public String getDriverClassName() {return driverClassName;}public void setDriverClassName(String driverClassName) {this.driverClassName = driverClassName;}public String getUrl() {return url;}public void setUrl(String url) {this.url = url;}public String getUsername() {return username;}public void setUsername(String username) {this.username = username;}public String getPassword() {return password;}public void setPassword(String password) {this.password = password;}public int getInitialSize() {return initialSize;}public void setInitialSize(int initialSize) {this.initialSize = initialSize;}public int getMinIdle() {return minIdle;}public void setMinIdle(int minIdle) {this.minIdle = minIdle;}public int getMaxActive() {return maxActive;}public void setMaxActive(int maxActive) {this.maxActive = maxActive;}public long getTimeBetweenEvictionRunsMillis() {return timeBetweenEvictionRunsMillis;}public void setTimeBetweenEvictionRunsMillis(long timeBetweenEvictionRunsMillis) {this.timeBetweenEvictionRunsMillis = timeBetweenEvictionRunsMillis;}public long getMinEvictableIdleTimeMillis() {return minEvictableIdleTimeMillis;}public void setMinEvictableIdleTimeMillis(long minEvictableIdleTimeMillis) {this.minEvictableIdleTimeMillis = minEvictableIdleTimeMillis;}public String getValidationQuery() {return validationQuery;}public void setValidationQuery(String validationQuery) {this.validationQuery = validationQuery;}public boolean isTestWhileIdle() {return testWhileIdle;}public void setTestWhileIdle(boolean testWhileIdle) {this.testWhileIdle = testWhileIdle;}public boolean isTestOnBorrow() {return testOnBorrow;}public void setTestOnBorrow(boolean testOnBorrow) {this.testOnBorrow = testOnBorrow;}public boolean isTestOnReturn() {return testOnReturn;}public void setTestOnReturn(boolean testOnReturn) {this.testOnReturn = testOnReturn;}public boolean isPoolPreparedStatements() {return poolPreparedStatements;}public void setPoolPreparedStatements(boolean poolPreparedStatements) {this.poolPreparedStatements = poolPreparedStatements;}public int getMaxPoolPreparedStatementPerConnectionSize() {return maxPoolPreparedStatementPerConnectionSize;}public void setMaxPoolPreparedStatementPerConnectionSize(int maxPoolPreparedStatementPerConnectionSize) {this.maxPoolPreparedStatementPerConnectionSize = maxPoolPreparedStatementPerConnectionSize;}public String getFilters() {return filters;}public void setFilters(String filters) {this.filters = filters;}public String getConnectionProperties() {return connectionProperties;}public void setConnectionProperties(String connectionProperties) {this.connectionProperties = connectionProperties;} } package com.imooc.mq.config.database;import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.support.PropertySourcesPlaceholderConfigurer; import org.springframework.jdbc.datasource.DataSourceTransactionManager; import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.annotation.EnableTransactionManagement;import javax.sql.DataSource; import java.sql.SQLException;import com.alibaba.druid.pool.DruidDataSource; /*** @Title: DruidDataSourceConfig* @Description: Druid數據源初始化** EnableTransactionManagement 開啟事務* @date 2019/1/2214:35*/@Configuration @EnableTransactionManagement public class DruidDataSourceConfig {private static Logger logger = LoggerFactory.getLogger(com.imooc.mq.config.database.DruidDataSourceConfig.class);//注入數據源配置信息 @Autowiredprivate DruidDataSourceSettings druidSettings;public static String DRIVER_CLASSNAME;@Beanpublic static PropertySourcesPlaceholderConfigurer propertyConfigure() {return new PropertySourcesPlaceholderConfigurer();}/*** 創建DataSource* @return* @throws SQLException*/@Beanpublic DataSource dataSource() throws SQLException {DruidDataSource ds = new DruidDataSource();ds.setDriverClassName(druidSettings.getDriverClassName());DRIVER_CLASSNAME = druidSettings.getDriverClassName();ds.setUrl(druidSettings.getUrl());ds.setUsername(druidSettings.getUsername());ds.setPassword(druidSettings.getPassword());ds.setInitialSize(druidSettings.getInitialSize());ds.setMinIdle(druidSettings.getMinIdle());ds.setMaxActive(druidSettings.getMaxActive());ds.setTimeBetweenEvictionRunsMillis(druidSettings.getTimeBetweenEvictionRunsMillis());ds.setMinEvictableIdleTimeMillis(druidSettings.getMinEvictableIdleTimeMillis());ds.setValidationQuery(druidSettings.getValidationQuery());ds.setTestWhileIdle(druidSettings.isTestWhileIdle());ds.setTestOnBorrow(druidSettings.isTestOnBorrow());ds.setTestOnReturn(druidSettings.isTestOnReturn());ds.setPoolPreparedStatements(druidSettings.isPoolPreparedStatements());ds.setMaxPoolPreparedStatementPerConnectionSize(druidSettings.getMaxPoolPreparedStatementPerConnectionSize());ds.setFilters(druidSettings.getFilters());ds.setConnectionProperties(druidSettings.getConnectionProperties());logger.info(" druid datasource config : {} ", ds);return ds;}/*** 開啟事務* @return* @throws Exception*/@Beanpublic PlatformTransactionManager transactionManager() throws Exception {DataSourceTransactionManager txManager = new DataSourceTransactionManager();txManager.setDataSource(dataSource());return txManager;} } package com.imooc.mq.config.database;import org.apache.ibatis.session.SqlSessionFactory; import org.mybatis.spring.SqlSessionFactoryBean; import org.mybatis.spring.SqlSessionTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.core.io.support.PathMatchingResourcePatternResolver; import org.springframework.core.io.support.ResourcePatternResolver;import javax.sql.DataSource;/*** @Title: MybatisDataSourceConfig* @Description: 整合mybatis和Druid* @date 2019/1/2214:39*/ @Configuration public class MybatisDataSourceConfig {@Autowiredprivate DataSource dataSource;@Bean(name="sqlSessionFactory")public SqlSessionFactory sqlSessionFactoryBean() {SqlSessionFactoryBean bean = new SqlSessionFactoryBean();bean.setDataSource(dataSource);// 添加XML目錄ResourcePatternResolver resolver = new PathMatchingResourcePatternResolver();try {bean.setMapperLocations(resolver.getResources("classpath:mapping/*.xml"));SqlSessionFactory sqlSessionFactory = bean.getObject();sqlSessionFactory.getConfiguration().setCacheEnabled(Boolean.TRUE);return sqlSessionFactory;} catch (Exception e) {throw new RuntimeException(e);}}@Beanpublic SqlSessionTemplate sqlSessionTemplate(SqlSessionFactory sqlSessionFactory) {return new SqlSessionTemplate(sqlSessionFactory);} } package com.imooc.mq.config.database;import org.mybatis.spring.mapper.MapperScannerConfigurer; import org.springframework.boot.autoconfigure.AutoConfigureAfter; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; /*** @Title: MybatisMapperScanerConfig* @Description: 掃碼Mybatis* @AutoConfigureAfter(MybatisDataSourceConfig.class) 先加載數據源類，再加載該類* @date 2019/1/2214:43*/ @Configuration @AutoConfigureAfter(MybatisDataSourceConfig.class) public class MybatisMapperScanerConfig {@Beanpublic MapperScannerConfigurer mapperScannerConfigurer() {MapperScannerConfigurer mapperScannerConfigurer = new MapperScannerConfigurer();mapperScannerConfigurer.setSqlSessionFactoryBeanName("sqlSessionFactory");mapperScannerConfigurer.setBasePackage("com.imooc.mq.mapper");return mapperScannerConfigurer;} }

?

package com.imooc.mq.config.task;import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.scheduling.annotation.EnableScheduling; import org.springframework.scheduling.annotation.SchedulingConfigurer; import org.springframework.scheduling.config.ScheduledTaskRegistrar;import java.util.concurrent.Executor; import java.util.concurrent.Executors; /*** @Title: TaskSchedulerConfig* @Description: 定時任務配置* @date 2019/1/2214:46*/ @Configuration @EnableScheduling //啟動定時任務 public class TaskSchedulerConfig implements SchedulingConfigurer {@Overridepublic void configureTasks(ScheduledTaskRegistrar taskRegistrar) {taskRegistrar.setScheduler(taskScheduler());}/*** 定時任務線程池* @return*/@Bean(destroyMethod = "shutdown")public Executor taskScheduler(){return Executors.newScheduledThreadPool(100);} } package com.imooc.mq.constant;/*** @Title: Constans* @Description: 常量* @date 2019/1/2214:50*/ public class Constans {/*** 發送中*/public static final String ORDER_SENDING = "0";/*** 發送成功*/public static final String ORDER_SEND_SUCCESS = "1";/*** 發送失敗*/public static final String ORDER_SEND_FAILURE = "2";/*** 分鐘超時單位：min*/public static final int ORDER_TIMEOUT = 1; }

---

?相應的mapper接口和mapper.xml文件配置

package com.imooc.mq.mapper;import com.imooc.mq.entity.BrokerMessageLog; import org.apache.ibatis.annotations.Param; import org.springframework.stereotype.Repository;import java.util.Date; import java.util.List;/*** @Title: BrokerMessageLogMapper* @Description: 消息記錄接口* @date 2019/1/2214:45*/ @Repository public interface BrokerMessageLogMapper {/*** 查詢消息狀態為0(發送中) 且已經超時的消息集合* @return*/List<BrokerMessageLog> query4StatusAndTimeoutMessage();/*** 重新發送統計count發送次數 +1* @param messageId* @param updateTime*/void update4ReSend(@Param("messageId")String messageId, @Param("updateTime") Date updateTime);/*** 更新最終消息發送結果 成功 or 失敗* @param messageId* @param status* @param updateTime*/void changeBrokerMessageLogStatus(@Param("messageId")String messageId, @Param("status")String status, @Param("updateTime")Date updateTime);int insertSelective(BrokerMessageLog record); } ------------------------------------------------------------------ package com.imooc.mq.mapper;import com.imooc.mq.entity.Order; import org.springframework.stereotype.Repository;/*** @Title: OrderMapper* @Description: 訂單接口* @date 2019/1/2214:45*/ @Repository public interface OrderMapper {int insert(Order record);int deleteByPrimaryKey(Integer id);int insertSelective(Order record);Order selectByPrimaryKey(Integer id);int updateByPrimaryKeySelective(Order record);int updateByPrimaryKey(Order record); } <?xml version="1.0" encoding="UTF-8" ?> <!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd" > <mapper namespace="com.imooc.mq.mapper.BrokerMessageLogMapper" ><resultMap id="BaseResultMap" type="com.imooc.mq.entity.BrokerMessageLog" ><id column="message_id" property="messageId" jdbcType="VARCHAR" /><result column="message" property="message" jdbcType="VARCHAR" /><result column="try_count" property="tryCount" jdbcType="INTEGER" /><result column="status" property="status" jdbcType="VARCHAR" /><result column="next_retry" property="nextRetry" jdbcType="TIMESTAMP" /><result column="create_time" property="createTime" jdbcType="TIMESTAMP" /><result column="update_time" property="updateTime" jdbcType="TIMESTAMP" /></resultMap><sql id="Base_Column_List" >message_id, message, try_count, status, next_retry, create_time, update_time</sql><select id="selectByPrimaryKey" resultMap="BaseResultMap" parameterType="java.lang.String" >select<include refid="Base_Column_List" />from broker_message_logwhere message_id = #{messageId,jdbcType=VARCHAR}</select><delete id="deleteByPrimaryKey" parameterType="java.lang.String" >delete from broker_message_logwhere message_id = #{messageId,jdbcType=VARCHAR}</delete><insert id="insert" parameterType="com.imooc.mq.entity.BrokerMessageLog" >insert into broker_message_log (message_id, message, try_count,status, next_retry, create_time,update_time)values (#{messageId,jdbcType=VARCHAR}, #{message,jdbcType=VARCHAR}, #{tryCount,jdbcType=INTEGER},#{status,jdbcType=VARCHAR}, #{nextRetry,jdbcType=TIMESTAMP}, #{createTime,jdbcType=TIMESTAMP},#{updateTime,jdbcType=TIMESTAMP})</insert><insert id="insertSelective" parameterType="com.imooc.mq.entity.BrokerMessageLog" >insert into broker_message_log<trim prefix="(" suffix=")" suffixOverrides="," ><if test="messageId != null" >message_id,</if><if test="message != null" >message,</if><if test="tryCount != null" >try_count,</if><if test="status != null" >status,</if><if test="nextRetry != null" >next_retry,</if><if test="createTime != null" >create_time,</if><if test="updateTime != null" >update_time,</if></trim><trim prefix="values (" suffix=")" suffixOverrides="," ><if test="messageId != null" >#{messageId,jdbcType=VARCHAR},</if><if test="message != null" >#{message,jdbcType=VARCHAR},</if><if test="tryCount != null" >#{tryCount,jdbcType=INTEGER},</if><if test="status != null" >#{status,jdbcType=VARCHAR},</if><if test="nextRetry != null" >#{nextRetry,jdbcType=TIMESTAMP},</if><if test="createTime != null" >#{createTime,jdbcType=TIMESTAMP},</if><if test="updateTime != null" >#{updateTime,jdbcType=TIMESTAMP},</if></trim></insert><update id="updateByPrimaryKeySelective" parameterType="com.imooc.mq.entity.BrokerMessageLog" >update broker_message_log<set ><if test="message != null" >message = #{message,jdbcType=VARCHAR},</if><if test="tryCount != null" >try_count = #{tryCount,jdbcType=INTEGER},</if><if test="status != null" >status = #{status,jdbcType=VARCHAR},</if><if test="nextRetry != null" >next_retry = #{nextRetry,jdbcType=TIMESTAMP},</if><if test="createTime != null" >create_time = #{createTime,jdbcType=TIMESTAMP},</if><if test="updateTime != null" >update_time = #{updateTime,jdbcType=TIMESTAMP},</if></set>where message_id = #{messageId,jdbcType=VARCHAR}</update><update id="updateByPrimaryKey" parameterType="com.imooc.mq.entity.BrokerMessageLog" >update broker_message_logset message = #{message,jdbcType=VARCHAR},try_count = #{tryCount,jdbcType=INTEGER},status = #{status,jdbcType=VARCHAR},next_retry = #{nextRetry,jdbcType=TIMESTAMP},create_time = #{createTime,jdbcType=TIMESTAMP},update_time = #{updateTime,jdbcType=TIMESTAMP}where message_id = #{messageId,jdbcType=VARCHAR}</update><select id="query4StatusAndTimeoutMessage" resultMap="BaseResultMap"><![CDATA[select message_id, message, try_count, status, next_retry, create_time, update_timefrom broker_message_log bmlwhere status = '0'and next_retry <= sysdate()]]></select><update id="update4ReSend" >update broker_message_log bmlset bml.try_count = bml.try_count + 1,bml.update_time = #{updateTime, jdbcType=TIMESTAMP}where bml.message_id = #{messageId,jdbcType=VARCHAR}</update><update id="changeBrokerMessageLogStatus" >update broker_message_log bmlset bml.status = #{status,jdbcType=VARCHAR},bml.update_time = #{updateTime, jdbcType=TIMESTAMP}where bml.message_id = #{messageId,jdbcType=VARCHAR}</update></mapper> ------------------------------------------------------------- <?xml version="1.0" encoding="UTF-8" ?> <!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd" > <mapper namespace="com.imooc.mq.mapper.OrderMapper" ><resultMap id="BaseResultMap" type="com.imooc.mq.entity.Order" ><id column="id" property="id" jdbcType="INTEGER" /><result column="name" property="name" jdbcType="VARCHAR" /><result column="message_id" property="messageId" jdbcType="VARCHAR" /></resultMap><sql id="Example_Where_Clause" ><where ><foreach collection="oredCriteria" item="criteria" separator="or" ><if test="criteria.valid" ><trim prefix="(" suffix=")" prefixOverrides="and" ><foreach collection="criteria.criteria" item="criterion" ><choose ><when test="criterion.noValue" >and ${criterion.condition}</when><when test="criterion.singleValue" >and ${criterion.condition} #{criterion.value}</when><when test="criterion.betweenValue" >and ${criterion.condition} #{criterion.value} and #{criterion.secondValue}</when><when test="criterion.listValue" >and ${criterion.condition}<foreach collection="criterion.value" item="listItem" open="(" close=")" separator="," >#{listItem}</foreach></when></choose></foreach></trim></if></foreach></where></sql><sql id="Update_By_Example_Where_Clause" ><where ><foreach collection="example.oredCriteria" item="criteria" separator="or" ><if test="criteria.valid" ><trim prefix="(" suffix=")" prefixOverrides="and" ><foreach collection="criteria.criteria" item="criterion" ><choose ><when test="criterion.noValue" >and ${criterion.condition}</when><when test="criterion.singleValue" >and ${criterion.condition} #{criterion.value}</when><when test="criterion.betweenValue" >and ${criterion.condition} #{criterion.value} and #{criterion.secondValue}</when><when test="criterion.listValue" >and ${criterion.condition}<foreach collection="criterion.value" item="listItem" open="(" close=")" separator="," >#{listItem}</foreach></when></choose></foreach></trim></if></foreach></where></sql><sql id="Base_Column_List" >id, name, message_id</sql><select id="selectByPrimaryKey" resultMap="BaseResultMap" parameterType="java.lang.Integer" >select<include refid="Base_Column_List" />from t_orderwhere id = #{id,jdbcType=INTEGER}</select><delete id="deleteByPrimaryKey" parameterType="java.lang.Integer" >delete from t_orderwhere id = #{id,jdbcType=INTEGER}</delete><insert id="insert" parameterType="com.imooc.mq.entity.Order" >insert into t_order (id, name, message_id)values (#{id,jdbcType=INTEGER}, #{name,jdbcType=VARCHAR}, #{messageId,jdbcType=VARCHAR})</insert><insert id="insertSelective" parameterType="com.imooc.mq.entity.Order" >insert into t_order<trim prefix="(" suffix=")" suffixOverrides="," ><if test="id != null" >id,</if><if test="name != null" >name,</if><if test="messageId != null" >message_id,</if></trim><trim prefix="values (" suffix=")" suffixOverrides="," ><if test="id != null" >#{id,jdbcType=INTEGER},</if><if test="name != null" >#{name,jdbcType=VARCHAR},</if><if test="messageId != null" >#{messageId,jdbcType=VARCHAR},</if></trim></insert><update id="updateByExampleSelective" parameterType="map" >update t_order<set ><if test="record.id != null" >id = #{record.id,jdbcType=INTEGER},</if><if test="record.name != null" >name = #{record.name,jdbcType=VARCHAR},</if><if test="record.messageId != null" >message_id = #{record.messageId,jdbcType=VARCHAR},</if></set><if test="_parameter != null" ><include refid="Update_By_Example_Where_Clause" /></if></update><update id="updateByExample" parameterType="map" >update t_orderset id = #{record.id,jdbcType=INTEGER},name = #{record.name,jdbcType=VARCHAR},message_id = #{record.messageId,jdbcType=VARCHAR}<if test="_parameter != null" ><include refid="Update_By_Example_Where_Clause" /></if></update><update id="updateByPrimaryKeySelective" parameterType="com.imooc.mq.entity.Order" >update t_order<set ><if test="name != null" >name = #{name,jdbcType=VARCHAR},</if><if test="messageId != null" >message_id = #{messageId,jdbcType=VARCHAR},</if></set>where id = #{id,jdbcType=INTEGER}</update><update id="updateByPrimaryKey" parameterType="com.imooc.mq.entity.Order" >update t_orderset name = #{name,jdbcType=VARCHAR},message_id = #{messageId,jdbcType=VARCHAR}where id = #{id,jdbcType=INTEGER}</update> </mapper>

?

package com.imooc.mq.entity;import java.util.Date;/*** @Title: BrokerMessageLog* @Description: 消息記錄* @date 2019/1/2214:29*/ public class BrokerMessageLog {private String messageId;private String message;private Integer tryCount;private String status;private Date nextRetry;private Date createTime;private Date updateTime;public BrokerMessageLog() {}public BrokerMessageLog(String messageId, String message, Integer tryCount, String status, Date nextRetry, Date createTime, Date updateTime) {this.messageId = messageId;this.message = message;this.tryCount = tryCount;this.status = status;this.nextRetry = nextRetry;this.createTime = createTime;this.updateTime = updateTime;}public String getMessageId() {return messageId;}public void setMessageId(String messageId) {this.messageId = messageId;}public String getMessage() {return message;}public void setMessage(String message) {this.message = message;}public Integer getTryCount() {return tryCount;}public void setTryCount(Integer tryCount) {this.tryCount = tryCount;}public String getStatus() {return status;}public void setStatus(String status) {this.status = status;}public Date getNextRetry() {return nextRetry;}public void setNextRetry(Date nextRetry) {this.nextRetry = nextRetry;}public Date getCreateTime() {return createTime;}public void setCreateTime(Date createTime) {this.createTime = createTime;}public Date getUpdateTime() {return updateTime;}public void setUpdateTime(Date updateTime) {this.updateTime = updateTime;} } -------------------------------------------------------------- package com.imooc.mq.entity;import java.io.Serializable;/*** @Title: Order* @Description: 訂單* @date 2019/1/2210:18*/ public class Order implements Serializable {private String id;private String name;//存儲消息發送的唯一標識private String messageId;public Order() {}public Order(String id, String name, String messageId) {this.id = id;this.name = name;this.messageId = messageId;}public String getId() {return id;}public void setId(String id) {this.id = id;}public String getName() {return name;}public void setName(String name) {this.name = name;}public String getMessageId() {return messageId;}public void setMessageId(String messageId) {this.messageId = messageId;}}

?

---

?

現在開始按照設計思路寫實現代碼：

　　1、首先我們把最核心了生產者寫好，生產者組成有基本的消息投遞，和監聽

package com.imooc.mq.producer;import com.imooc.mq.constant.Constans; import com.imooc.mq.entity.Order; import com.imooc.mq.mapper.BrokerMessageLogMapper; import org.slf4j.Logger; import org.slf4j.LoggerFactory; 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;import java.util.Date;/*** @Title: RabbitOrderSender* @Description: 消息發送* @date 2019/1/2214:52*/ @Component public class RabbitOrderSender {private static Logger logger = LoggerFactory.getLogger(RabbitOrderSender.class);@Autowiredprivate RabbitTemplate rabbitTemplate;@Autowiredprivate BrokerMessageLogMapper brokerMessageLogMapper;/*** Broker應答后，會調用該方法區獲取應答結果*/final RabbitTemplate.ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {logger.info("correlationData："+correlationData);String messageId = correlationData.getId();if (ack){//如果返回成功，則進行更新brokerMessageLogMapper.changeBrokerMessageLogStatus(messageId, Constans.ORDER_SEND_SUCCESS,new Date());}else {//失敗進行操作：根據具體失敗原因選擇重試或補償等手段logger.error("異常處理"+cause);}}};/*** 發送消息方法調用: 構建自定義對象消息* @param order* @throws Exception*/public void sendOrder(Order order) throws Exception {// 通過實現 ConfirmCallback 接口，消息發送到 Broker 后觸發回調，確認消息是否到達 Broker 服務器，也就是只確認是否正確到達 Exchange 中 rabbitTemplate.setConfirmCallback(confirmCallback);//消息唯一IDCorrelationData correlationData = new CorrelationData(order.getMessageId());rabbitTemplate.convertAndSend("order-exchange1", "order.ABC", order, correlationData);} }

　　2、將定時任務邏輯寫好

package com.imooc.mq.task;import com.imooc.mq.constant.Constans; import com.imooc.mq.entity.BrokerMessageLog; import com.imooc.mq.entity.Order; import com.imooc.mq.mapper.BrokerMessageLogMapper; import com.imooc.mq.producer.RabbitOrderSender; import com.imooc.mq.utils.FastJsonConvertUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component;import java.util.Date; import java.util.List;/*** @Title: RetryMessageTasker* @Description: 定時任務* @date 2019/1/2215:45*/ @Component public class RetryMessageTasker {private static Logger logger = LoggerFactory.getLogger(RetryMessageTasker.class);@Autowiredprivate RabbitOrderSender rabbitOrderSender;@Autowiredprivate BrokerMessageLogMapper brokerMessageLogMapper;/*** 定時任務*/@Scheduled(initialDelay = 5000, fixedDelay = 10000)public void reSend(){logger.info("-----------定時任務開始-----------");//抽取消息狀態為0且已經超時的消息集合List<BrokerMessageLog> list = brokerMessageLogMapper.query4StatusAndTimeoutMessage();list.forEach(messageLog -> {//投遞三次以上的消息if(messageLog.getTryCount() >= 3){//更新失敗的消息brokerMessageLogMapper.changeBrokerMessageLogStatus(messageLog.getMessageId(), Constans.ORDER_SEND_FAILURE, new Date());} else {// 重試投遞消息，將重試次數遞增brokerMessageLogMapper.update4ReSend(messageLog.getMessageId(), new Date());Order reSendOrder = FastJsonConvertUtil.convertJSONToObject(messageLog.getMessage(), Order.class);try {rabbitOrderSender.sendOrder(reSendOrder);} catch (Exception e) {e.printStackTrace();logger.error("-----------異常處理-----------");}}});}}

　　3、寫好消費者的邏輯，直接用上一篇中的消費者代碼，修改對應的exchange、queue、路由key就好

package com.imooc.mq.consumer;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-queue1",declare = "true"),exchange = @Exchange(name = "order-exchange1",declare = "true",type = "topic"),key = "order.ABC"))@RabbitHandlerpublic 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);} }

　　4、業務邏輯

package com.imooc.mq.service;import com.imooc.mq.constant.Constans; import com.imooc.mq.entity.BrokerMessageLog; import com.imooc.mq.entity.Order; import com.imooc.mq.mapper.BrokerMessageLogMapper; import com.imooc.mq.mapper.OrderMapper; import com.imooc.mq.producer.RabbitOrderSender; import com.imooc.mq.utils.DateUtils; import com.imooc.mq.utils.FastJsonConvertUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service;import java.util.Date;/*** @Title: OrderService* @Description: 業務實現* @date 2019/1/2215:41*/ @Service public class OrderService {private static Logger logger = LoggerFactory.getLogger(OrderService.class);@Autowiredprivate OrderMapper orderMapper;@Autowiredprivate BrokerMessageLogMapper brokerMessageLogMapper;@Autowiredprivate RabbitOrderSender rabbitOrderSender;public void createOrder(Order order) {try {// 使用當前時間當做訂單創建時間（為了模擬一下簡化）Date orderTime = new Date();// 插入業務數據 orderMapper.insert(order);// 插入消息記錄表數據BrokerMessageLog brokerMessageLog = new BrokerMessageLog();// 消息唯一ID brokerMessageLog.setMessageId(order.getMessageId());// 保存消息整體 轉為JSON 格式存儲入庫 brokerMessageLog.setMessage(FastJsonConvertUtil.convertObjectToJSON(order));// 設置消息狀態為0 表示發送中brokerMessageLog.setStatus("0");// 設置消息未確認超時時間窗口為 一分鐘 brokerMessageLog.setNextRetry(DateUtils.addMinutes(orderTime, Constans.ORDER_TIMEOUT));brokerMessageLog.setCreateTime(new Date());brokerMessageLog.setUpdateTime(new Date());brokerMessageLogMapper.insertSelective(brokerMessageLog);// 發送消息 rabbitOrderSender.sendOrder(order);} catch (Exception e) {logger.error("訂單業務異常{}",e);}} }

　　5、測試

/*** 測試訂單創建*/@Testpublic void createOrder(){Order order = new Order();order.setId("201901228");order.setName("測試訂單");order.setMessageId(System.currentTimeMillis() + "$" + UUID.randomUUID().toString());try {orderService.createOrder(order);} catch (Exception e) {e.printStackTrace();}}

　　先啟動消費者服務、再啟動生產者服務讓定時任務跑起來，最后啟動測試方法。消息被消費成功后，日志記錄狀態被修改為1。測試消息重投的話需要制造一些異常情況，比如修改消費者端跟exchange，生產者找不到該交互機，拿不到回調，就會重試投遞。

?

轉載于:https://www.cnblogs.com/zhangbLearn/p/10304976.html

總結

以上是生活随笔為你收集整理的SpringBoot整合RabbitMQ-消息可靠性投递的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。