单channel监听RabbitMQ消息队列，多线程处理消息任务的实现

ゝ一纸荒年。 2022-05-31 01:29 257阅读 0赞

项目上碰到一个java后台实现问题，底层依赖的以前开发团队的rabbitMQ的实现，该底层open了一个channel，并一直保持着对该channel的监听，在这个基础上我需要实现的就是一直监听该channel，并将接受到的消息交给其他线程去处理，保证监听主线程不被阻塞。需求大致是这样的。

下面看下我的具体实现：

一、主线程监听消息队列，并将接受的消息交给其他线程去处理，并且处理消息的其他线程数量有限制。这里考虑用concurrent包里的semaphore，信号量可以认为是一定数量num的许可证，主线程将消息和一张许可证交给一个子线程去处理，子线程处理完了（不管成功或失败）都要将许可证上交给主线程，供下一个子线程使用，这样就可能出现两种情况：1）许可证富裕，主线程继续将消息交给子线程去消费；2）许可证已被多个子线程占用，还未交回主线程手里。这样主线程就阻塞了。这个玩意刚好满足我的业务需求。看下代码：

private int threadNum = 10;
    private Semaphore semaphore = new Semaphore(threadNum);
    private Executor executor = Executors.newCachedThreadPool();
    private int allowedChildThreads = 10;
    
    public void doJob() {
    		Logger.info("begin consuming!", new Object[0]);
    		MessageClientFactory.getClient().consume("xxxQueue", true, new MessageConsumer() {
    			@Override
    			public void consume(String message) {
    				Logger.info(message, new Object[0]);
    				
    				try {
    					JSONObject messageObject = JSONObject.parseObject(message);
    					
    					String jobType = messageObject.getString("job_type"); //任务类型
    					String jobID = messageObject.getString("job_id"); //任务ID
    					Date jobSubmitTime = messageObject.getDate("job_submit_time"); //任务提交时间
    					String params = messageObject.getString("params"); //任务指标参数
    					
    					//构建线程间信息传递对象
    					Map<String, String> threadDelivery = new HashMap<>();
    					
    					threadDelivery.put("job_type", jobType);
    					threadDelivery.put("job_id", jobID);
    					threadDelivery.put("params", params);
    					
    					semaphore.acquire(); //许可证减少一张
    					executor.execute(new MessageSplitToMany(threadDelivery, semaphore, executor, allowedChildThreads)); //将消费信息和许可证一并交给子线程MessageSplitToMany，他处理完释放许可证
    				} catch (Exception e) {
    					Logger.error(e, "消费主线程发生错误", new Object[0]);
    				}
    			}
    		});
    		
    	}

二、MessageSplitToMany子线程进一步将消息拆分并交给一定数量的孙线程处理（用semaphore限制并发数），在这些孙线程未全部处理完成前，MessageSplitToMany子线程阻塞。这里CountDownLatch正好满足需求。

/**
     * 一条消息分解成多个指标线程并行计算
     * @author fujian
     *
     */
    public class MessageSplitToMany implements Runnable {
    	
    	private Map<String, String> threadDelivery;
    	private Semaphore semaphore;
    	private Executor executor;
    	private int allowedChildThreads;
    	
    	public MessageSplitToMany(Map<String, String> threadDelivery, Semaphore semaphore, Executor executor, int allowedChildThreads) {
    		this.threadDelivery = threadDelivery;
    		this.semaphore = semaphore;
    		this.executor = executor;
    		this.allowedChildThreads = allowedChildThreads;
    	}
    	@Override
    	public void run() {
    		try {
    			String params = threadDelivery.get("params");
    			Logger.info(params, new Object[0]);
    			JSONObject multiIdxParams = JSONObject.parseObject(params);
    			//对并发处理单个指标计算的工作线程数量有一定的约束,所允许的工作线程数量不能超过设定值
    			if(allowedChildThreads > multiIdxParams.size()) {
    				allowedChildThreads = multiIdxParams.size();
    							}
    			
    			Semaphore innerSemaphore = new Semaphore(allowedChildThreads);
    			CountDownLatch countDownLatch = new CountDownLatch(multiIdxParams.size());
    			for(String idxName : multiIdxParams.keySet()) {
    				String idxParams = multiIdxParams.getString(idxName);
    				
    				Map<String, Object> childThreadDelivery = new LinkedHashMap<>();
    				childThreadDelivery.put("job_id", threadDelivery.get("job_id"));
    				childThreadDelivery.put("index_id", DigestUtils.md5Hex(idxName + idxParams));
    				childThreadDelivery.put("job_type", threadDelivery.get("job_type"));
    				childThreadDelivery.put("idx_name", idxName);
    				childThreadDelivery.put("params", idxParams);
    				
    				innerSemaphore.acquire(); //孙信号量控制子线程中的并发数
    				executor.execute(new IdxCalAndSaveThread(countDownLatch, innerSemaphore, childThreadDelivery));//将countDownLatch及孙信号量传给IdxCalAndSaveThread孙线程，由它countdown及释放孙信号量
    			}
    			
    			countDownLatch.await(); //在message中所有指标计算完成前主线程阻塞，直至countdown减为0
    			
    			Logger.info("-----" + Thread.currentThread().getName() + "consumer thread finished the all idxCals Cal-----");
    			
    		} catch (Exception e) {
    			Logger.error(e, "并发处理消息过程发生错误", new Object[0]);
    		} finally {
    			semaphore.release(); //释放子信号量，这个信号量是从主线程传递进来的
    		}
    	}
    
    }

三、孙线程是最底层的处理线程，其最终要释放孙信号量及对countDownLatch减一，

/**
     * 指标计算及存储
     * @author fujian
     *
     */
    public class IdxCalAndSaveThread implements Runnable {
    	
    	private Semaphore innerSemaphore;
    	private CountDownLatch countDownLatch;
    	private Map<String, Object> threadDelivery;
    	
    	public IdxCalAndSaveThread(CountDownLatch countDownLatch, Semaphore innerSemaphore, Map<String, Object> threadDelivery) {
    		this.countDownLatch = countDownLatch;
    		this.innerSemaphore = innerSemaphore;
    		this.threadDelivery = threadDelivery;
    	}
    	
    	
    	@Override
    	public void run() {
    		try {
    				String idxName = threadDelivery.get("idx_name").toString();
    				
    				String params = threadDelivery.get("params").toString();
    				threadDelivery.remove("params");
    				
    				threadDelivery.put("start_time", System.currentTimeMillis());
    				
    				//调用indexCal 层指标计算api，计算指标
    				//考虑idxName为idxCal_01
    				Object idx;
    				if(idxName.contains("_")) {
    					String idxCalName = idxName.substring(0, idxName.lastIndexOf("_"));
    					idx = IdxCalHandler.handle(idxCalName, params);
    				} else {
    					idx = IdxCalHandler.handle(idxName, params);
    				}
    
    				threadDelivery.put("end_time", System.currentTimeMillis());
    				threadDelivery.put("idx_result", idx);
    				Logger.info(idxName + "cal is finished!", new Object[0]);
    				
    				//save to mongo
    				saveToDB(threadDelivery);
    				Logger.info(idxName + "cal result had saved into mongo", new Object[0]);
    			
    		} catch (Exception e) {
    			Logger.error(e, "单个指标计算过程发生错误", new Object[0]);
    		} finally { //最终要释放孙信号量及countdown减一
    			innerSemaphore.release();
    			countDownLatch.countDown();
    		}
    	}
    }