前面我們介紹了RabbitMQ的基本概念,RabbitMQ基礎概念詳細介紹。在這裡我們做一個簡單的例子進行快速入門。
建立Spring Boot項目 引入依賴包
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>1.5.3.RELEASE</version>
<relativePath /> <!-- lookup parent from repository -->
</parent>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
在啟動類上添加啟動MQ的注解
@EnableRabbit
添加配置
# Rabbitmq
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.virtual-host=test
spring.rabbitmq.addresses=192.168.35.128:5672
#spring.rabbitmq.addresses=192.168.35.128:5672,192.168.35.129:5672,192.168.35.130:5672
spring.rabbitmq.connection-timeout=50000
#rabbitmq listetner
# 消費者最小數量
spring.rabbitmq.listener.concurrency=10
# 消費者最大數量
spring.rabbitmq.listener.max-concurrency=20
# 消息的确認模式
spring.rabbitmq.listener.acknowledge-mode=MANUAL
# 每一次發送到消費者的消息數量,它應該大于或等于事務大小(如果使用)。
spring.rabbitmq.listener.prefetch=10
# 消費者端的重試
spring.rabbitmq.listener.retry.enabled=true
#rabbitmq publisher
# 生産者端的重試
spring.rabbitmq.template.retry.enabled=true
#開啟發送消息到exchange确認機制
spring.rabbitmq.publisher-confirms=true
#開啟發送消息到exchange但是exchange沒有和隊列綁定的确認機制
spring.rabbitmq.publisher-returns=true
RabbitMQ 所有配置參考:
# RABBIT (RabbitProperties)
spring.rabbitmq.addresses= # Comma-separated list of addresses to which the client should connect.
spring.rabbitmq.cache.channel.checkout-timeout= # Number of milliseconds to wait to obtain a channel if the cache size has been reached.
spring.rabbitmq.cache.channel.size= # Number of channels to retain in the cache.
spring.rabbitmq.cache.connection.mode=CHANNEL # Connection factory cache mode.
spring.rabbitmq.cache.connection.size= # Number of connections to cache.
spring.rabbitmq.connection-timeout= # Connection timeout, in milliseconds; zero for infinite.
spring.rabbitmq.dynamic=true # Create an AmqpAdmin bean.
spring.rabbitmq.host=localhost # RabbitMQ host.
spring.rabbitmq.listener.acknowledge-mode= # Acknowledge mode of container.
spring.rabbitmq.listener.auto-startup=true # Start the container automatically on startup.
spring.rabbitmq.listener.concurrency= # Minimum number of consumers.
spring.rabbitmq.listener.default-requeue-rejected= # Whether or not to requeue delivery failures; default `true`.
spring.rabbitmq.listener.max-concurrency= # Maximum number of consumers.
spring.rabbitmq.listener.prefetch= # Number of messages to be handled in a single request. It should be greater than or equal to the transaction size (if used).
spring.rabbitmq.listener.retry.enabled=false # Whether or not publishing retries are enabled.
spring.rabbitmq.listener.retry.initial-interval=1000 # Interval between the first and second attempt to deliver a message.
spring.rabbitmq.listener.retry.max-attempts=3 # Maximum number of attempts to deliver a message.
spring.rabbitmq.listener.retry.max-interval=10000 # Maximum interval between attempts.
spring.rabbitmq.listener.retry.multiplier=1.0 # A multiplier to apply to the previous delivery retry interval.
spring.rabbitmq.listener.retry.stateless=true # Whether or not retry is stateless or stateful.
spring.rabbitmq.listener.transaction-size= # Number of messages to be processed in a transaction. For best results it should be less than or equal to the prefetch count.
spring.rabbitmq.password= # Login to authenticate against the broker.
spring.rabbitmq.port=5672 # RabbitMQ port.
spring.rabbitmq.publisher-confirms=false # Enable publisher confirms.
spring.rabbitmq.publisher-returns=false # Enable publisher returns.
spring.rabbitmq.requested-heartbeat= # Requested heartbeat timeout, in seconds; zero for none.
spring.rabbitmq.ssl.enabled=false # Enable SSL support.
spring.rabbitmq.ssl.key-store= # Path to the key store that holds the SSL certificate.
spring.rabbitmq.ssl.key-store-password= # Password used to access the key store.
spring.rabbitmq.ssl.trust-store= # Trust store that holds SSL certificates.
spring.rabbitmq.ssl.trust-store-password= # Password used to access the trust store.
spring.rabbitmq.ssl.algorithm= # SSL algorithm to use. By default configure by the rabbit client library.
spring.rabbitmq.template.mandatory=false # Enable mandatory messages.
spring.rabbitmq.template.receive-timeout=0 # Timeout for `receive()` methods.
spring.rabbitmq.template.reply-timeout=5000 # Timeout for `sendAndReceive()` methods.
spring.rabbitmq.template.retry.enabled=false # Set to true to enable retries in the `RabbitTemplate`.
spring.rabbitmq.template.retry.initial-interval=1000 # Interval between the first and second attempt to publish a message.
spring.rabbitmq.template.retry.max-attempts=3 # Maximum number of attempts to publish a message.
spring.rabbitmq.template.retry.max-interval=10000 # Maximum number of attempts to publish a message.
spring.rabbitmq.template.retry.multiplier=1.0 # A multiplier to apply to the previous publishing retry interval.
spring.rabbitmq.username= # Login user to authenticate to the broker.
spring.rabbitmq.virtual-host= # Virtual host to use when connecting to the broker.
聲明隊列
@Configuration
@ConditionalOnBean({RabbitTemplate.class})
public class RabbitConfig {
/**
* 方法rabbitAdmin的功能描述:動态聲明queue、exchange、routing
*
* @param connectionFactory
* @return
* @author : yuhao.wang
*/
@Bean
public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
// 發放獎勵隊列交換機
DirectExchange exchange = new DirectExchange(RabbitConstants.MQ_EXCHANGE_SEND_AWARD);
//聲明發送優惠券的消息隊列(Direct類型的exchange)
Queue couponQueue = queue(RabbitConstants.QUEUE_NAME_SEND_COUPON);
rabbitAdmin.declareQueue(couponQueue);
rabbitAdmin.declareExchange(exchange);
rabbitAdmin.declareBinding(BindingBuilder.bind(couponQueue).to(exchange).with(RabbitConstants.MQ_ROUTING_KEY_SEND_COUPON));
return rabbitAdmin;
}
public Queue queue(String name) {
// 是否持久化
boolean durable = true;
// 僅建立者可以使用的私有隊列,斷開後自動删除
boolean exclusive = false;
// 當所有消費用戶端連接配接斷開後,是否自動删除隊列
boolean autoDelete = false;
return new Queue(name, durable, exclusive, autoDelete, args);
}
}
在這裡我們申明了一個RabbitConstants.QUEUE_NAME_SEND_COUPON隊列,并聲明了一個DirectExchange 類型的交換器,通過Bind将隊列、交換機和路由RabbitConstants.MQ_ROUTING_KEY_SEND_COUPON的關系進行綁定。
消息的生産者
/**
* Rabbit 發送消息
*
* @author yuhao.wang
*/
@Service
public class RabbitSender implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback, InitializingBean {
private final Logger logger = LoggerFactory.getLogger(RabbitSender.class);
/**
* Rabbit MQ 用戶端
*/
@Autowired
private RabbitTemplate rabbitTemplate;
/**
* 系統配置
*/
@Autowired
private SystemConfig systemConfig;
/**
* 發送MQ消息
*
* @param exchangeName 交換機名稱
* @param routingKey 路由名稱
* @param message 發送消息體
*/
public void sendMessage(String exchangeName, String routingKey, Object message) {
// 擷取CorrelationData對象
CorrelationData correlationData = this.correlationData(message);
rabbitTemplate.convertAndSend(exchangeName, routingKey, message, correlationData);
}
/**
* 用于實作消息發送到RabbitMQ交換器後接收ack回調。
* 如果消息發送确認失敗就進行重試。
*
* @param correlationData
* @param ack
* @param cause
*/
@Override
public void confirm(org.springframework.amqp.rabbit.support.CorrelationData correlationData, boolean ack, String cause) {
// 消息回調确認失敗處理
if (!ack) {
// 這裡以做消息的從發等處理
logger.info("消息發送失敗,消息ID:{}", correlationData.getId());
} else {
logger.info("消息發送成功,消息ID:{}", correlationData.getId());
}
}
/**
* 用于實作消息發送到RabbitMQ交換器,但無相應隊列與交換器綁定時的回調。
* 基本上來說線程不可能出現這種情況,除非手動将已經存在的隊列删掉,否則在測試階段肯定能測試出來。
*/
@Override
public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
logger.error("MQ消息發送失敗,replyCode:{}, replyText:{},exchange:{},routingKey:{},消息體:{}",
replyCode, replyText, exchange, routingKey, JSON.toJSONString(message.getBody()));
// TODO 儲存消息到資料庫
}
/**
* 消息相關資料(消息ID)
*
* @param message
* @return
*/
private CorrelationData correlationData(Object message) {
return new CorrelationData(UUID.randomUUID().toString(), message);
}
@Override
public void afterPropertiesSet() throws Exception {
rabbitTemplate.setConfirmCallback(this);
rabbitTemplate.setReturnCallback(this);
}
}
ConfirmCallback和ReturnCallback
在這個裡我們主要實作了ConfirmCallback和ReturnCallback兩個接口。這兩個接口主要是用來發送消息後回調的。因為rabbit發送消息是隻管發,至于發沒發成功,發送方法不管。
- ConfirmCallback:當消息成功到達exchange的時候觸發的ack回調。
- ReturnCallback:當消息成功到達exchange,但是沒有隊列與之綁定的時候觸發的ack回調。基本上來說線上不可能出現這種情況,除非手動将已經存在的隊列删掉,否則在測試階段肯定能測試出來。
如果使用RabbitMQ的ConfirmCallback和ReturnCallback模式必須将下面兩個開關打開,否則将不生效:
# 生産者端的重試
spring.rabbitmq.template.retry.enabled=true
#開啟發送消息到exchange确認機制
spring.rabbitmq.publisher-confirms=true
消息的發送方式
- rabbitTemplate.send(message); //發消息,參數類型為org.springframework.amqp.core.Message
- rabbitTemplate.convertAndSend(object); //轉換并發送消息。 将參數對象轉換為org.springframework.amqp.core.Message後發送,這個是異步的。消息是否發送成功需要用到ConfirmCallback和ReturnCallback回調函數類确認。
- rabbitTemplate.convertSendAndReceive(message) //轉換并發送消息,且等待消息者傳回響應消息。這是一個RPC方法,當發送消息過後,該方法會一直阻塞在哪裡等待傳回結果,直到請求逾時。可以通過配置spring.rabbitmq.template.reply-timeout來配置逾時時間。
消息的消費者
/**
* 發放優惠券的MQ處理
*
* @author yuhao.wang
*/
public class SendMessageListener {
private final Logger logger = LoggerFactory.getLogger(SendMessageListener.class);
@RabbitListener(queues = RabbitConstants.QUEUE_NAME_SEND_COUPON)
public void process(SendMessage sendMessage, Channel channel, Message message) throws Exception {
try {
// 參數校驗
Assert.notNull(sendMessage, "sendMessage 消息體不能為NULL");
logger.info("處理MQ消息");
// 确認消息已經消費成功
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
} catch (Exception e) {
logger.error("MQ消息處理異常,消息ID:{},消息體:{}", message.getMessageProperties().getCorrelationIdString(),
JSON.toJSONString(sendMessage), e);
// 拒絕目前消息,并把消息傳回原隊列
channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
}
}
}
使用 @RabbitListener注解,并在注解上指定你要監聽的隊列名稱,這樣子消費者就聲明好了。這裡有兩點要注意一下:
- 監聽消息的參數(如上面的sendMessage參數)一定要和發送的時候相比對,也可以不使用sendMessage參數,直接在Message參數裡面擷取消息體。
- 如果有傳回資訊,直接return就好(rabbitTemplate.convertSendAndReceive()方法就會有傳回值)。
消息的确認和拒絕
消息确認
我看可以調用Channel類中的basicAck方法進行消息确認,方法定義如下:
void basicAck(long deliveryTag, boolean multiple) throws IOException;
- deliveryTag:消息編号。
- multiple:是否确認多條消息。false,确認目前這條消息;true,确認deliveryTag編号以前的所有消息。
消息拒絕
拒絕消息可以使用Channel中的basicReject或者basicNack方法,basicReject隻能拒絕一條消息,basicNack可以拒絕多條消息。
void basicReject(long deliveryTag, boolean requeue) throws IOException;
void basicNack(long deliveryTag, boolean multiple, boolean requeue)
throws IOException;
- deliveryTag:消息編号。
- requeue:是否将消息放回隊列。true,将消息放回隊列;false,丢棄消息。
- multiple:是否拒絕多條消息。false,拒絕目前這條消息(這樣子就和basicReject一樣了);true,拒絕deliveryTag編号以前的所有消息。
源碼
https://github.com/wyh-spring-ecosystem-student/spring-boot-student/tree/releases