rabbitmq批量处理 -欧洲杯足彩官网

我们通过spring-amqp操作rabbitmq是极其简单的，消息的生产者和消费者只需要如下配置：

客户端（生产者）：connectionfactory、queue、exchange、messageconverter、rabbittemplate。

服务端（消费者）：connectionfactory、queue、exchange、messageconverter、listenercontainer。

如果消息堆积严重，我们可以通过两种方式来处理消息，一种是在服务端开启监听多线程服务（concurrency="10"），另一种是让消息批量出队列。

开启多线程的配置示例如下：

批量出队列的示例如下：

客户端（消息生产者）

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() {
    			@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 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() {
			@override
			public string call() throws exception {
				execute();
				return null;
			}			
		});
	}
	
	private void execute(){		
		while (true) {
			rabbittemplate.execute(new channelcallback() {
				@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 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) {
			}
		}
	}
}