spring+mybatsi FIFO(先进先出)队列，解决并发问题

墨蓝 2023-10-17 10:30 5阅读 0赞

业务场景：最近负责公司的“邮件系统”的研发工作，公司其他的业务系统都需要调用“邮件服务接口”，导致该邮件服务接口，并发量比较大。

解决思路：

![Center][]

1、web 初始化时，同时需要初始化系统全局的FIFO队列。

2、java FIFO队列的选择，必须是线程安全队列，我这里使用的是BlockingQueue。

3、邮件发送失败，需要重新添到FIFO队列中。

4、spring 定时任务框架，实时处理FIFO队列中的任务。

核心示例代码:

1、web 初始化 FIFO 队列

package com.settlement.initialization;
    
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.LinkedBlockingQueue;
    
    import javax.servlet.ServletContext;
    
    import org.springframework.beans.factory.InitializingBean;
    import org.springframework.web.context.ServletContextAware;
    
    public class InitFIFOListener implements InitializingBean, ServletContextAware{
    	
    	public BlockingQueue<Object> queue = new  LinkedBlockingQueue<Object>();
    
    	public void setServletContext(ServletContext context) {
    		// TODO Auto-generated method stub
    		
    	}
    
    	public void afterPropertiesSet() throws Exception {
    		// TODO Auto-generated method stub
    		
    	}
    
    	public BlockingQueue<Object> getQueue() {
    		return queue;
    	}
    
    	public void setQueue(BlockingQueue<Object> queue) {
    		this.queue = queue;
    	}
    	
    	
    	
    	
    
    }

spring-mybatis.xml 配置文件

<bean id="initFIFO" class="com.settlement.initialization.InitFIFOListener">
        </bean>

2、邮件服务接口。

package com.settlement.controller;
    
    import java.math.BigDecimal;
    import java.util.Queue;
    import java.util.concurrent.BlockingQueue;
    
    import javax.servlet.http.HttpServletRequest;
    
    import org.springframework.stereotype.Controller;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RequestMethod;
    import org.springframework.web.bind.annotation.ResponseBody;
    
    import com.settlement.commons.base.BaseController;
    import com.settlement.commons.email.SendMailUtil;
    import com.settlement.commons.scan.SpringUtils;
    import com.settlement.initialization.InitFIFOListener;
    import com.settlement.model.AccountWallet;
    import com.taobao.api.internal.util.StringUtils;
    
    /**
     * 邮箱控制器Controller
     * 
     * @author zzg
     * @date 2017-02-10
     */
    @Controller
    @RequestMapping(value = "/email")
    public class EmailController extends BaseController {
    	private BlockingQueue<Object> queue = null;
    
    	/**
    	 * 针对业务系统高并发-----采用FIFO队列，进行解决
    	 * 
    	 * @return
    	 */
    	@RequestMapping(value = "/receiveemail.action", method = RequestMethod.POST)
    	@ResponseBody
    	public String receiveEmail(HttpServletRequest request) {
    
    		String result = "";
    		try {
    			//项目全局FIFO 队列
    			if (queue == null) {
    				InitFIFOListener listener = (InitFIFOListener) SpringUtils
    						.getContext().getBean("initFIFO");
    				queue = listener.queue;
    			}
    			System.out.println("队列大小："+queue.size());
    			String receiveMailAccount = request.getParameter("email") == null ? null
    					: request.getParameter("email").trim();
    			queue.offer(receiveMailAccount);
    
    		} catch (Exception e) {
    			result = e.getMessage();
    			return result;
    		}
    
    		return "success";
    	}
    
    	/**
    	 * 处理FIFO队列，需要发送的邮件信息
    	 * 
    	 * @return
    	 */
    	@RequestMapping(value = "/sendemail.action", method = RequestMethod.POST)
    	@ResponseBody
    	public String sendEmail() {
    
    		String result = "";
    		boolean isRunning = true;
    		try {
    			//项目全局FIFO 队列
    			if (queue == null) {
    				InitFIFOListener listener = (InitFIFOListener) SpringUtils
    						.getContext().getBean("initFIFO");
    				queue = listener.queue;
    			}
    			while (isRunning) {
    				System.out.println("队列大小："+queue.size());
    				System.out.println("正从队列获取数据...");
    				String data = (String) queue.poll();
    				if (null != data) {					
    					System.out.println("正在消费数据：" + data);
    					SendMailUtil.send_163(data);
    				} else {
    					// 超过2s还没数据，认为所有生产线程都已经退出，自动退出消费线程。
    					isRunning = false;
    				}
    			}
    
    		} catch (Exception e) {
    			result = e.getMessage();
    			return result;
    		} finally {
    			System.out.println("退出消费者线程！");
    		}
    
    		return "success";
    	}
    
    }

3、高并发测试，邮件服务接口。

package com.settlement.concurrent;
    
    import com.settlement.commons.utils.HttpRequest;
    
    public class EmailConcurrentTest {
    
    	/**
    	 * @param args
    	 */
    	public static void main(String[] args) {
    		// TODO Auto-generated method stub
    		//模拟调用邮件服务接口30次
    //		for(int i=0;i<30;i++){
    //			HttpRequest.sendPost("http://localhost:8080/Settlement/email/receiveemail.action", "email=2355540290@qq.com");
    //		}
    		//模拟定时任务，处理FIFO队列
    		HttpRequest.sendPost("http://localhost:8080/Settlement/email/sendemail.action", "");
    
    	}
    
    }

温馨提示： [Java 邮件发送参考文章][Java]  
  
4、效果截图：

待补充。

[Center]: https://img-blog.csdn.net/20170210203317152?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvemhvdXpoaXdlbmdhbmc=/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/Center
[Java]: http://blog.csdn.net/zhouzhiwengang/article/details/54973252