标签:out enable Fix read mis type mos 设置 date
前篇 消息队列客户端开发向导(基于 Spring 的 amqp 实现)
一、应答模式与重试机制
NONE 自动应答 即 autoAck 等于 true,rabbitmq 会自动把发送出去的消息置为确认,然后从内存/磁盘中删除,而不管消费者是否真正消费了消息。
MANUAL 需要手动 ACK/NACK
AUTO 根据是否抛出异常来决定 ACK/NACK
注:如果 应答模式为 AUTO,并且 retry.enabled 为 false,如果发生异常则会一直重试。
配置参考
spring: rabbitmq: host: 122.51.195.163 username: rabbitmq password: rabbitmq listener: type: simple simple: retry: enabled: true max-attempts: 1 max-concurrency: 100 prefetch: 26 batch-size: 1000 # 需要配合 receive-timeout 来使用 acknowledge-mode: auto template: receive-timeout: 50S # 50秒 超时
二、批处理
批量发送
配置批处理策略,当批完成时才发送消息给 rabbitmq
@Bean public BatchingRabbitTemplate batchingRabbitTemplate(ConnectionFactory connectionFactory) { BatchingStrategy strategy = new SimpleBatchingStrategy(1000, 25_000, 3_000); TaskScheduler scheduler = new ConcurrentTaskScheduler(); BatchingRabbitTemplate template = new BatchingRabbitTemplate(strategy, scheduler); template.setConnectionFactory(connectionFactory); return template; }
批处理发送消息
private final BatchingRabbitTemplate batchingRabbitTemplate; public void sendTask(TempMsgCbTask task) { batchingRabbitTemplate.convertAndSend(TEMP_MSG_CB_EXCHANGE, task.getLevel(), task); }
批量消费
默认批数据会自动 debatch 成单个。为了一次消费多个,需要设置 batchListener 为 true
@Bean(name = "rabbitListenerContainerFactory") @ConditionalOnProperty(prefix = "spring.rabbitmq.listener", name = "type", havingValue = "simple", matchIfMissing = true) public SimpleRabbitListenerContainerFactory simpleRabbitListenerContainerFactory( SimpleRabbitListenerContainerFactoryConfigurer configurer, ConnectionFactory connectionFactory) { SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory(); factory.setBatchListener(true); // if consumerBatchEnabled is true , deBatchingEnabled will be true. refer https://docs.spring.io/spring-amqp/docs/2.2.7.RELEASE/reference/html/#receiving-batch // factory.setConsumerBatchEnabled(true); configurer.configure(factory, connectionFactory); return factory; }
批量消费消息
@RabbitListener(queues = TEMP_MSG_CB_LOG_QUEUE_1, priority = "1") public void logQueue1(List<TempMsgCbTask> in) { log.info("Message read from logQueue1 msgId"); List<TemplateSendCollect> sendCollects = tempPushStore.updateCbDetailAndLog(in); templateSendLogService.doUpdate(sendCollects); }
当数据迟迟没有满足批处理大小 batch-size 时,如果达到超时时间 receive-timeout 将会提前消费。
233
消息队列客户端开发向导二(基于 Spring 的 amqp 实现)
标签:out enable Fix read mis type mos 设置 date
原文地址:https://www.cnblogs.com/lemos/p/12989900.html