几个概念说明：
Broker：简单来说就是消息队列服务器实体。
Exchange：消息交换机，它指定消息按什么规则，路由到哪个队列。
Queue：消息队列载体，每个消息都会被投入到一个或多个队列。
Binding：绑定，它的作用就是把exchange和queue按照路由规则绑定起来。
Routing Key：路由关键字，exchange根据这个关键字进行消息投递。
vhost：虚拟主机，一个broker里可以开设多个vhost，用作不同用户的权限分离。
producer：消息生产者，就是投递消息的程序。
consumer：消息消费者，就是接受消息的程序。
channel：消息通道，在客户端的每个连接里，可建立多个channel，每个channel代表一个会话任务。
交换机路由的几种类型:
Direct Exchange:直接匹配,通过Exchange名称+RountingKey来发送与接收消息.
Fanout Exchange:广播订阅,向所有的消费者发布消息,但是只有消费者将队列绑定到该路由器才能收到消息,忽略Routing Key.
Topic Exchange：主题匹配订阅,这里的主题指的是RoutingKey,RoutingKey可以采用通配符,如:*或#，RoutingKey命名采用.来分隔多个词,只有消息这将队列绑定到该路由器且指定RoutingKey符合匹配规则时才能收到消息;
Headers Exchange:消息头订阅,消息发布前,为消息定义一个或多个键值对的消息头,然后消费者接收消息同时需要定义类似的键值对请求头:(如:x-mactch=all或者x_match=any)，只有请求头与消息头匹配,才能接收消息,忽略RoutingKey.
默认的exchange:如果用空字符串去声明一个exchange，那么系统就会使用”amq.direct”这个exchange，我们创建一个queue时,默认的都会有一个和新建queue同名的routingKey绑定到这个默认的exchange上去
安装Erland
http://www.erlang.org/downloads
安装RabbitMQ
https://www.rabbitmq.com/download.html
开启RabbitMQ服务
执行rabbitmq-plugins enable rabbitmq_management命令，开启Web管理插件
重启RabbitMQ服务
Web地址
http://localhost:15672/
默认用户名和密码：guest
一、引入springboot和rabbitmq的依赖
<dependency> <groupId>org.quartz-schedulergroupId> <artifactId>quartzartifactId> <version>2.3.0version> dependency>
二、新增application.properties对rabbimq的配置信息
spring.application.name=springboot-rabbitmq
spring.rabbitmq.host=116.255.193.36
spring.rabbitmq.port=5672
spring.rabbitmq.username=scrm
spring.rabbitmq.password=scrm
spring.rabbitmq.publisher-cOnfirms=true
spring.rabbitmq.publisher-returns=true
spring.rabbitmq.virtual-host=scrm
spring.rabbitmq.listener.simple.acknowledge-mode=manual
#最小消息监听线程数
spring.rabbitmq.listener.cOncurrency=2
#最大消息监听线程数
spring.rabbitmq.listener.max-cOncurrency=2
三、公共设置类
1、队列、消息交换机，路由关键字公共枚举类
package cloud.app.prod.home.rabbitmq; /** * Author : YongBo Xie * File Name: RabbitMqEnum.java * Created Date: 2018年3月28日 上午10:32:02 * Modified Date: 2018年3月28日 上午10:32:02 * Version: 1.0 */ public class RabbitMqEnum { /** * describe: 定义队列名称 **/ public enum QueueName { MARKETING_ACTIVITIE_QUEUE("marketingActivitieQueue", "营销活动队列"); private String code; private String name; QueueName(String code, String name) { this.code = code; this.name = name; } public String getCode() { return code; } public String getName() { return name; } } /** * describe: 定义交换机 **/ public enum Exchange { DIRECT_EXCHANGE("directExchange", "直连交换机"), FANOUT_EXCHANGE("fanoutExchange", "扇形交换机"), TOPIC_EXCHANGE("topicExchange", "主题交换机"), HEADERS_EXCHANGE("headersExchange", "首部交换机"); private String code; private String name; Exchange(String code, String name) { this.code = code; this.name = name; } public String getCode() { return code; } public String getName() { return name; } } /** * describe: 定义routing_key **/ public enum QueueKey { MARKETING_ACTIVITIE_DIRECT("marketingActivitie", "营销活动key"), MARKETING_ACTIVITIE_TOPIC_01("*.marketingActivitie.*", "营销活动key"), MARKETING_ACTIVITIE_TOPIC_02("marketingActivitie.#", "营销活动key"); private String code; private String name; QueueKey(String code, String name) { this.code = code; this.name = name; } public String getCode() { return code; } public String getName() { return name; } } }
2、数据连接配置类
package cloud.app.prod.home.rabbitmq; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.connection.ConnectionFactory; 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.Configuration; /** * Author : YongBo Xie * File Name: RabbitConfig.java * Created Date: 2018年3月28日 下午6:41:17 * Modified Date: 2018年3月28日 下午6:41:17 * Version: 1.0 */ @Configuration @ConfigurationProperties(prefix = "spring.rabbitmq") public class RabbitConfig { @Value("${spring.rabbitmq.host}") private String addresses; @Value("${spring.rabbitmq.port}") private int port; @Value("${spring.rabbitmq.username}") private String username; @Value("${spring.rabbitmq.password}") private String password; @Value("${spring.rabbitmq.publisher-confirms}") private Boolean publisherConfirms; @Value("${spring.rabbitmq.publisher-returns}") private Boolean publisherReturns; @Value("${spring.rabbitmq.virtual-host}") private String virtualHost; // 构建mq实例工厂 @Bean public ConnectionFactory connectionFactory() { CachingConnectionFactory connectionFactory = new CachingConnectionFactory(); connectionFactory.setAddresses(addresses); connectionFactory.setPort(port); connectionFactory.setUsername(username); connectionFactory.setPassword(password); connectionFactory.setPublisherConfirms(publisherConfirms); connectionFactory.setVirtualHost(virtualHost); connectionFactory.setPublisherReturns(publisherReturns); return connectionFactory; } }
3、生产者类
package cloud.app.prod.home.rabbitmq; import org.apache.log4j.Logger; import org.springframework.amqp.core.Message; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.core.RabbitTemplate.ConfirmCallback; import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback; import org.springframework.amqp.rabbit.support.CorrelationData; import org.springframework.context.annotation.Bean; import org.springframework.stereotype.Component; /** * Author : YongBo Xie * File Name: RabbitMqSender.java * Created Date: 2018年3月30日 上午10:48:36 * Modified Date: 2018年3月30日 上午10:48:36 * Version: 1.0 */ @Component public class RabbitMqSender { private static Logger logger = Logger.getLogger(RabbitMqSender.class); @Bean public RabbitTemplate messageRabbitTemplate(ConnectionFactory connectionFactory) { final RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory); rabbitTemplate.setConfirmCallback(new ConfirmCallback() { /** * 回调 * @param correlationData 消息唯一标识 * @param ack 确认结果 * @param cause 失败原因 */ @Override public void confirm(CorrelationData correlationData, boolean ack, String cause) { logger.info("消息唯一标识："+correlationData); logger.info("确认结果："+ack); logger.info("失败原因："+cause); } }); rabbitTemplate.setReturnCallback(new ReturnCallback() { /** * 用于实现消息发送到RabbitMQ交换器，但无相应队列与交换器绑定时的回调 */ @Override public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) { logger.info(message.getMessageProperties().getCorrelationIdString() + " 发送失败"); } }); return rabbitTemplate; } }
四、个例
1、初始化队列、消息交换机，并把队列绑定到消息交换机
京公网安备 11010802041100号