1、引入依赖
compile 'org.springframework.boot:spring-boot-starter-amqp'
2、配置文件：
spring.rabbitmq.host=10.0.60.121
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=admin
spring.rabbitmq.virtual-host=/
3、配置代码
import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.PropertySource;
import org.springframework.context.annotation.Scope;
import com.wow.common.mq.MqMetaConstant;
@Configuration
@PropertySource("classpath:/rabbitmq.properties")
public class RabbitMqConfiguration {
@Value("${spring.rabbitmq.host}")private String springRabbitmqHost;
@Value("${spring.rabbitmq.port}") private Integer springRabbitmqPort;
@Value("${spring.rabbitmq.username}") private String springRabbitmqUsername;
@Value("${spring.rabbitmq.password}") private String springRabbitmqPassword;
@Value("${spring.rabbitmq.virtual-host}") private String springRabbitmqVirtualHost;
@Bean public ConnectionFactory connectionFactory() {
CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
//防止信道缓存不够造成消息丢失，官方推荐100可完全避免此丢失消息情况
connectionFactory.setChannelCacheSize(100);
connectionFactory.setHost(springRabbitmqHost);
connectionFactory.setPort(springRabbitmqPort);
connectionFactory.setUsername(springRabbitmqUsername);
connectionFactory.setPassword(springRabbitmqPassword);
connectionFactory.setVirtualHost(springRabbitmqVirtualHost);
connectionFactory.setChannelCacheSize(100);
// 必须要设置 connectionFactory.setPublisherConfirms(true);
return connectionFactory;
}
/** *
* 用于恢复交换机，队列
*
* @return
*/
@Bean
public AmqpAdmin amqpAdmin() {
return new RabbitAdmin(connectionFactory());
}
// 必须是prototype类型
@Bean
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
public RabbitTemplate rabbitTemplate() {
RabbitTemplate template = new RabbitTemplate(connectionFactory());
return template;
}
/**
* FanoutExchange: 将消息分发到所有的绑定队列，无routingkey的概念
* HeadersExchange：通过添加属性key-value匹配
* DirectExchange:按照routingkey分发到指定队列
* TopicExchange:多关键字匹配
* @return
*/
@Bean
public DirectExchange defaultExchange() {
return new DirectExchange(MqMetaConstant.EXCHANGE_EMAIL);
}
@Bean
public FanoutExchange adminMessageExchange() {
return new FanoutExchange(MqMetaConstant.EXCHANGE_ADMIN_MESSAGE); }
@Bean
public Queue queue() {
return new Queue(MqMetaConstant.QUEUE_EMAIL, true); // 队列持久 }
@Bean
public Queue adminMessageQueue() {
return new Queue(MqMetaConstant.QUEUE_ADMIN_MESSAGE, true); // 队列持久 }
@Bean
public Binding binding() {
return BindingBuilder.bind(queue()).to(defaultExchange()).with(MqMetaConstant.ROUTINGKEY_EMAIL); }
@Bean
public Binding adminMessageBinding() {
return BindingBuilder.bind(adminMessageQueue()).to(adminMessageExchange()); }
}
import java.util.List;
import java.util.UUID;
import com.wow.common.mq.vo.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import com.wow.common.util.CollectionUtil;
import com.wow.common.util.JsonUtil;
/**
* 发送消息到rabbitmq
* @author wangming@wowdsgn.com
* @date 2017年11月6日 下午5:41:58
*/
@Component
public class MqProducer implements RabbitTemplate.ConfirmCallback {
private static final Logger logger=LoggerFactory.getLogger(MqProducer.class);
private RabbitTemplate rabbitTemplate;
/** * 构造方法注入 */
@Autowired
public MqProducer(RabbitTemplate rabbitTemplate) {
this.rabbitTemplate = rabbitTemplate;
rabbitTemplate.setConfirmCallback(this);
//rabbitTemplate如果为单例的话，那回调就是最后设置的内容 rabbitTemplate.setReturnCallback((message,replyCode,replyText,exchange,routingKey)->{ logger.info("没有找到任何匹配的队列！"+ "message:"+message+ ",replyCode:"+replyCode+ ",replyText:"+replyText+ ",exchange:"+exchange+ ",routingKey:"+routingKey); }); }
/**
* 发送邮件消息
*
* @param emailVO
*/
public void sendEmail(EmailVO emailVO) {
if(emailVO==null) { logger.warn("email对象为空，拒绝发送邮件消息"); return ; }
CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());
String message=JsonUtil.pojo2Json(emailVO);
logger.info("开始发送mq邮件消息：id:{}，内容：{}，",correlationId,message); rabbitTemplate.convertAndSend(MqMetaConstant.EXCHANGE_EMAIL, MqMetaConstant.ROUTINGKEY_EMAIL,message , correlationId); }
/**
* 发送后台消息
* @param messageList
*/
public void sendAdminMessage(AdminMessageVO messageVO) {
CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());
String message=JsonUtil.pojo2Json(messageVO);
logger.info("开始发送后台消息：id:{}，内容：{}，",correlationId,message); rabbitTemplate.convertAndSend(MqMetaConstant.EXCHANGE_ADMIN_MESSAGE, "",message , correlationId); }
/** * 回调 */
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
if (ack) {
logger.info("mq消息发送成功---确认");
} else {
logger.info("mq消息发送失败:id:{}，cause：{}" , correlationData,cause);
}
}
}
网友评论