rabbitmq批量处理
程序员文章站
2022-07-04 11:37:46
...
我们通过spring-amqp操作rabbitmq是极其简单的，消息的生产者和消费者只需要如下配置：
客户端（生产者）：connectionFactory、queue、exchange、messageConverter、RabbitTemplate。
服务端（消费者）：connectionFactory、queue、exchange、messageConverter、listenerContainer。
如果消息堆积严重，我们可以通过两种方式来处理消息，一种是在服务端开启监听多线程服务（concurrency="10"），另一种是让消息批量出队列。
开启多线程的配置示例如下：
<rabbit:listener-container connection-factory="connectionFactory" acknowledge="auto" requeue-rejected="false"
concurrency="10" message-converter="jsonMessageConverter" >
<rabbit:listener ref="tradeListener" method="listen" queues="queue_trade_repay" />
</rabbit:listener-container>
批量出队列的示例如下：
客户端（消息生产者）
import java.math.BigDecimal;
import org.junit.Test;
import org.springframework.beans.factory.annotation.Autowired;
import com.rd.account.domain.AccountLog;
import com.rd.ifaes.mq.producer.RabbitProducer;
import com.rd.ifaes.web.BaseTest;
/**
* 消息生产者
* @author lihua
* @since 2018-04-08
*
*/
public class Producer extends BaseTest{
// @Autowired
// private RabbitTemplate rabbitTemplate;
//这里对rabbitTemplate做了简单的封装，您可以直接使用rabbitTemplate
@Autowired
private RabbitProducer rabbitProducer;
private static final String queueName = "ACCOUNT_LOG_BATCH"; //MqConstant.ROUTING_KEY_ACCOUNT_LOG_BATCH;
@Test
public void main() {
for (int i = 0; i < 512; i++) {
AccountLog log = new AccountLog("001", "001", "asdf", BigDecimal.valueOf(i), "remark"+i);
rabbitProducer.send(queueName, log);
// rabbitTemplate.convertAndSend(queueName, "hello" + i);
}
}
}
服务端（消息消费者）
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.core.ChannelCallback;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import com.alibaba.fastjson.JSONObject;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.QueueingConsumer;
/**
* 消息消费者
* @author lihua
* @since 2018-04-08
*
*/
public class Consumer extends BaseTest{
private static final Logger LOGGER = LoggerFactory.getLogger(Consumer.class);
@Autowired
private RabbitTemplate rabbitTemplate;
@Autowired
private AccountLogService accountLogService;
private static final String queueName = "ACCOUNT_LOG_BATCH"; //MqConstant.ROUTING_KEY_ACCOUNT_LOG_BATCH;
private static final int BATCH_SIZE = 100;
@Test
public void consumer() {
while (true) {
rabbitTemplate.execute(new ChannelCallback<String>() {
@Override
public String doInRabbit(Channel channel) throws Exception {
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
try {
final AMQP.Queue.DeclareOk ok = channel.queueDeclare(queueName, true, false, false, null);
int messageCount = ok.getMessageCount();
LOGGER.info("run consumer {}, msg count {}", sdf.format(new Date()), messageCount);
if (messageCount == 0) {
return null;
}
List<AccountLog> list = new ArrayList<>();
channel.basicQos(BATCH_SIZE);
QueueingConsumer queueingConsumer = new QueueingConsumer(channel);
LOGGER.info("channel id {}", Integer.toHexString(System.identityHashCode(channel)));
final String inConsumerTag = "test consumer" + sdf.format(new Date());
channel.basicConsume(queueName, false, inConsumerTag, queueingConsumer);
long messageId = -1;
int dealedCount = 0;
int i = BATCH_SIZE;
while (i-- > 0) {
QueueingConsumer.Delivery delivery = queueingConsumer.nextDelivery(BATCH_SIZE);
if (delivery == null) {
break;
}
String msg = new String(delivery.getBody());
AccountLog log = JSONObject.parseObject(msg, AccountLog.class);
list.add(log);
messageId = delivery.getEnvelope().getDeliveryTag();
LOGGER.info("get message {} delivery id {}", msg, messageId);
dealedCount++;
if (dealedCount % 5 == 0) {
channel.basicAck(messageId, true);
LOGGER.info("batch ack message id =>{}", messageId);
messageId = -1;
}
}
if (messageId > 0) {
channel.basicAck(messageId, true);
LOGGER.info("last to ack message id =>{}", messageId);
}
// 日志入库
accountLogService.saveBatch(list);
} finally {
LOGGER.info("consumer done {}", sdf.format(new Date()));
}
channel.abort();
return null;
}
});
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
}
}
}
}
import org.junit.runner.RunWith;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations={"classpath:spring-context.xml"})
public abstract class BaseTest {
}
补一个服务端真实案例：
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import javax.annotation.PostConstruct;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.core.ChannelCallback;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Component;
import com.alibaba.fastjson.JSONObject;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.QueueingConsumer;
@Component
@Lazy(value=false)
public class AccountLogBatchListener {
@Autowired
private RabbitTemplate rabbitTemplate;
@Autowired
private AccountLogService accountLogService;
private static final Logger LOGGER = LoggerFactory.getLogger(AccountLogBatchListener.class);
private static final String QUEUE_NAME = MqConstant.ROUTING_KEY_ACCOUNT_LOG_BATCH;
private static final ExecutorService executor = Executors.newFixedThreadPool(1);
private static final int BATCH_SIZE = 100;
@PostConstruct
public void init(){
executor.submit(new Callable<String>() {
@Override
public String call() throws Exception {
execute();
return null;
}
});
}
private void execute(){
while (true) {
rabbitTemplate.execute(new ChannelCallback<String>() {
@Override
public String doInRabbit(Channel channel) throws Exception {
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
try {
final AMQP.Queue.DeclareOk ok = channel.queueDeclare(QUEUE_NAME, true, false, false, null);
int messageCount = ok.getMessageCount();
LOGGER.debug("accountLogBatchListener {}, msg count {}", sdf.format(new Date()), messageCount);
if (messageCount == 0) {
return null;
}
List<AccountLog> list = new ArrayList<>();
channel.basicQos(BATCH_SIZE);
QueueingConsumer queueingConsumer = new QueueingConsumer(channel);
LOGGER.debug("channel id {}", Integer.toHexString(System.identityHashCode(channel)));
final String inConsumerTag = "accountLogBatchListener {}" + sdf.format(new Date());
channel.basicConsume(QUEUE_NAME, false, inConsumerTag, queueingConsumer);
long messageId = -1;
int dealedCount = 0;
int i = BATCH_SIZE;
while (i-- > 0) {
QueueingConsumer.Delivery delivery = queueingConsumer.nextDelivery(BATCH_SIZE);
if (delivery == null) {
break;
}
String msg = new String(delivery.getBody());
AccountLog log = JSONObject.parseObject(msg, AccountLog.class);
list.add(log);
messageId = delivery.getEnvelope().getDeliveryTag();
LOGGER.info(" userId {}, delivery id {}", log.getUserId(), messageId);
dealedCount++;
if (dealedCount % 5 == 0) {
channel.basicAck(messageId, true);
LOGGER.debug("batch ack message id =>{}", messageId);
messageId = -1;
}
}
if (messageId > 0) {
channel.basicAck(messageId, true);
LOGGER.debug("last to ack message id =>{}", messageId);
}
// 日志入库
accountLogService.saveBatch(list);
} finally {
LOGGER.info("accountLogBatchListener done {}", sdf.format(new Date()));
}
channel.abort();
return null;
}
});
try {
Thread.sleep(10000);
} catch (InterruptedException e) {
}
}
}
}
推荐阅读
-
山药过敏了应该怎么处理
-
在一个form用一个SUBMIT(或button)分别提交到两个处理表单页面的代码
-
在修改准备发的批量美化select+可修改select时，在非IE下发现了几个问题
-
茶叶有保质期吗？放久的茶叶怎么处理最好
-
肉蟹怎么杀？简单几步就能将肉蟹处理好
-
首款支持5G的iPad Pro预计9月发布：5nm A14处理器、后置多摄
-
Intel Lakefield处理器官方近照：3D 5核心的奥妙
-
The Linely Group年度最佳手机处理器大奖：神U麒麟810夺冠
-
华擎发布支持64核霄龙处理器的主板：内置512MB显存 采用VGA接口
-
三星Galaxy A41跑分曝光：搭载联发科P65处理器