阿里MNS和RabitMQ双支持方案

ゝ一纸荒年。 2022-07-13 12:47 212阅读 0赞

# `认识阿里MNS` #

![SouthEast][]

### RabbitMQ 三种Exchange ###

Direct Exchange – 处理路由键

Fanout Exchange – 不处理路由键

Topic Exchange – 将路由键和某模式进行匹配

protected void initRabitMq(){
          String user = mqProperties.getProperty("rabbit.username");
          String pwd = mqProperties.getProperty("rabbit.pwd");
          String address = mqProperties.getProperty("rabbit.addresses");
    
          this.exchangeName = mqProperties.getProperty("rabbit.exchangeName");
          this.routingKey=mqProperties.getProperty("rabbit.routingKey");
          
          ExponentialBackOffPolicy policy = new ExponentialBackOffPolicy();
          policy.setInitialInterval(500);
          policy.setMaxInterval(5000);
          policy.setMultiplier(10);
          RetryTemplate retryTemplate = new RetryTemplate();
          retryTemplate.setBackOffPolicy(policy);
                
          CachingConnectionFactory factory = new CachingConnectionFactory();
          factory.setAddresses(address);
          factory.setUsername(user);
          factory.setPassword(pwd);
          
          RabbitTemplate rabbitTemplate = new RabbitTemplate();
          rabbitTemplate.setRetryTemplate(retryTemplate);
          rabbitTemplate.setConnectionFactory(factory);
    //    rabbitTemplate.setMessageConverter(new FastJsonMessageConverter());
          
          RabbitMqService rabmqService = new RabbitMqService();
          rabmqService.setRabbitTemplate(rabbitTemplate);
          mqService = rabmqService;
       }
       
       protected void initAliMnsMq(){
          String endpoint = mqProperties.getProperty("mns.accountendpoint");
          String accessId = mqProperties.getProperty("mns.accesskeyid");
          String accessSec = mqProperties.getProperty("mns.accesskeysecret");
                
          CloudAccount account = new CloudAccount(accessId,accessSec,endpoint);
          AliyunMnsService alimqService = new AliyunMnsService();
          alimqService.setMnsAccount(account);
          
          mqService = alimqService;
       }

配置：

#mq
    #if start mq sender
    mq.startSender=1
    #is start mq msg receiver
    mq.startConsumer=0
    
    #mq.type=rabbitmq
    mq.type=aliyunmns
    
    #queuename prefix
    mq.queueName=iuap-ys-queue
    
    #rabit,need when mq.type=rabbitmq
    rabbit.exchangeName=iuap-ys-exchange
    #rabbit routing key prefix
    rabbit.routingKey=iuap-ys-key
      
    rabbit.username=guest
    rabbit.pwd=guest
    rabbit.addresses=localhost:5672
    
    #mns,need when mq.type=aliyunmns
    #\u4e0b\u9762\u662f\u90a2\u52a0\u519b\u7684\u6d4b\u8bd5\u8d26\u53f7
    mns.accountendpoint=http://mns.cn.aliyuncs.com/
    mns.accesskeyid=000000000000
    mns.accesskeysecret=888888888888888888888888888888

queue队列和exception队列

/**
     * hr的消息队列消费者
     * @author Administrator
     *
     */
    public class HrMqMessageListener implements ApplicationListener<ContextRefreshedEvent>{
    	private static Logger logger = LoggerFactory.getLogger(HrMqMessageListener.class);
    	
    	private Properties mqProperties = null;	
    	private Properties hrRestProperties = null;	
    	private String mqType = null;
    	private boolean isRunning = false;
    	
    	private SimpleMessageListenerContainer rabbitContainer=null;
    	private CachingConnectionFactory rabbitFactory = null;
    	private AbstractMessageListener mnsListener = null;
    	private BpmEngineConfiguration bpmEngineConfiguration;
    	
    	public HrMqMessageListener() {
    		if(logger.isInfoEnabled())
    			logger.info("MqMessageListener init");
    	}
    	
    	//应该在构造之后执行
    	public void start(){
    		if(isRunning){
    			return;
    		}
    		
    		String startConsumer = hrRestProperties.getProperty("mq.startConsumer");
    		if(!"1".equals(startConsumer)){
    			logger.info("do not start hr mq consumer");
    			return;
    		}
    		
    		
    		logger.info("start hr mq consumer...");
    		
    		mqType = mqProperties.getProperty("mq.type");
    		if(IMqService.MQTYPE_RABBIT.equals(mqType)){
    			initRabitMq();
    			isRunning = true;
    		}else if(IMqService.MQTYPE_MNS.equals(mqType)){
    			initAliMnsMq();
    			isRunning = true;
    		}else{
    			logger.info("no mq type set");
    		}
    	}
    	
    	private String getLocalIp(){
    		InetAddress ia = null;
    		try {
    			ia = InetAddress.getLocalHost();
    		} catch (UnknownHostException e) {			
    			e.printStackTrace();
    			return null;
    		}
    		return ia.getHostAddress();
    	}
    	
    	public void stop(){
    		if(!isRunning)
    			return;
    		
    		if(IMqService.MQTYPE_RABBIT.equals(mqType)){
    			if(rabbitContainer!=null)
    				rabbitContainer.stop();
    			if(rabbitFactory!=null)
    				rabbitFactory.destroy();
    		}else if(IMqService.MQTYPE_MNS.equals(mqType)){
    			//暂无停止接口
    		}
    		isRunning=false;
    	}
    	
    	public Properties getMqProperties() {
    		return mqProperties;
    	}
    
    	public void setMqProperties(Properties mqProperties) {
    		this.mqProperties = mqProperties;
    	}
    
    	
    	public Properties getHrRestProperties() {
    		return hrRestProperties;
    	}
    
    	public void setHrRestProperties(Properties hrRestProperties) {
    		this.hrRestProperties = hrRestProperties;
    	}
    
    	protected void initRabitMq(){
    		String user = mqProperties.getProperty("rabbit.username");
    		String pwd = mqProperties.getProperty("rabbit.pwd");
    		String address = mqProperties.getProperty("rabbit.addresses");
    		String queueName = hrRestProperties.getProperty("mq_hrclouds_queueName");//队列名称从hrproperties中取
    				
    		rabbitFactory = new CachingConnectionFactory();
    		rabbitFactory.setAddresses(address);
    		rabbitFactory.setUsername(user);
    		rabbitFactory.setPassword(pwd);
    		
    		rabbitContainer = new SimpleMessageListenerContainer();
    		rabbitContainer.setConnectionFactory(rabbitFactory);
    		rabbitContainer.setMessageListener(new MessageListener(){
    			public void onMessage(Message message){
    				String msg;
    				try {
    					msg = new String(message.getBody(),"UTF-8");
    				} catch (UnsupportedEncodingException e) {
    					 throw new RuntimeException("Not support encoding: UTF-8");
    				}
    				consumeMessage(msg);
    			}
    		}); 
    		rabbitContainer.setQueueNames(queueName);
    		
    		rabbitContainer.start();//启动
    	}
    	
    	protected void initAliMnsMq(){
    		String enbpoint = mqProperties.getProperty("mns.accountendpoint");
    		String accessId = mqProperties.getProperty("mns.accesskeyid");
    		String accessSec = mqProperties.getProperty("mns.accesskeysecret");
    		String queueName = hrRestProperties.getProperty("mq_hrclouds_queueName");//队列名称从hrproperties中取
    		
    		CloudAccount account = new CloudAccount(accessId,accessSec,enbpoint);
    		
    		mnsListener = new AbstractMessageListener(queueName,account,5){
    			@Override
    			public void onMessage(com.aliyun.mns.model.Message message) {
    				String msg = message.getMessageBody();
    				consumeMessage(msg);
    			}
    		};
    		
    		String ublinkqueueName = hrRestProperties.getProperty("mq_ublink_queueName");//队列名称从hrproperties中取
    		//CloudAccount ubLinkaccount = new CloudAccount(accessId,accessSec,enbpoint);
    		new AbstractMessageListener(ublinkqueueName,account,5){
    			@Override
    			public void onMessage(com.aliyun.mns.model.Message message) {
    				String msg = message.getMessageBody();
    				consumUblinkeMessage(msg);
    			}
    		};
    	}
    	
    	
    	protected void consumUblinkeMessage(String msg){
    		if(logger.isDebugEnabled())
    			logger.debug("get new ublink msg:"+msg);
    		
    		try {
    			sendUbLinkNoticeMessage(msg);
    		} catch (Throwable e) {
    			consumeFailed(msg,e);
    		}
    	}
    	
    	protected void sendUbLinkNoticeMessage(String msg){
    		
    		String url = hrRestProperties.getProperty("ublink_rest_url").trim();
    		String token = hrRestProperties.getProperty("ublink_rest_token");
    		
    		HttpPost post = HttpUtils.createPost(url);
    		HttpUtils.setHttpHeader(post, "token", token);
    		
    		StringEntity se = new StringEntity(msg,"UTF-8");
    		se.setContentEncoding("UTF-8");
    		se.setContentType("application/json");  
    
    		System.out.println("ublink queue msg send to ublink rest."+msg+",url="+url);
    		
    		String resp = HttpUtils.httpPost(post, se);//true
    		if(logger.isDebugEnabled())
    			logger.debug("ublink rest post over. resp is {}",resp);
    	}
    
    	
    	/**
    	 * 消费消息
    	 * @param msg 消息体
    	 */
    	protected void consumeMessage(String msg){
    		if(logger.isDebugEnabled())
    			logger.debug("get new mq msg:"+msg);
    		
    		try {
    			sendNoticeMessage(msg);
    		} catch (Throwable e) {
    			consumeFailed(msg,e);
    		}
    	}
    	
    	protected void consumeFailed(String mqMsg,Throwable t){
    		String consumeError = t.getMessage();
    		try {
    			JSONObject data = JSON.parseObject(mqMsg);//
    			data.put("EXCEPTION", consumeError);
    			String json = JSON.toJSONString(data);
    			bpmEngineConfiguration.getMqSender().send(json, "exception");			
    		} catch (JSONException e1) {//解析报文json错误
    			logger.error("parse mq message from json to JSONObject exception."+e1.getMessage(),e1);
    			e1.printStackTrace();
    			
    			//对于服务解析成json的报文			
    			String notJsonMqMsg = mqMsg+"\r\nJSONException:"+e1.getMessage();
    			bpmEngineConfiguration.getMqSender().send(notJsonMqMsg, "exception");
    		}
    	}
    	
    	/**
    	 * 将消息推送给系统的接口
    	 * @param sendType
    	 * @param taskResp
    	 * @param exeResp
    	 */
    	protected void sendNoticeMessage(String msg){
    		
    		String url = hrRestProperties.getProperty("hrclouds_rest_url").trim();
    		String token = hrRestProperties.getProperty("hrclouds_rest_token");
    		
    		HttpPost post = HttpUtils.createPost(url);
    		HttpUtils.setHttpHeader(post, "token", token);
    		
    		StringEntity se = new StringEntity(msg,"UTF-8");
    		se.setContentEncoding("UTF-8");
    		se.setContentType("application/json");  
    
    		System.out.println("hr queue msg send to hr rest."+msg+",url="+url);
    		
    		String resp = HttpUtils.httpPost(post, se);//true
    		if(logger.isDebugEnabled())
    			logger.debug("hr rest post over. resp is {}",resp);
    	}
    
    	@Override
    	public void onApplicationEvent(ContextRefreshedEvent event) {
    		if(logger.isDebugEnabled())
    			logger.debug("spring context started over.");
    		
    		ApplicationContext context = event.getApplicationContext();
    		
    //		Context.getProcessEngineConfiguration();//得到的是null,原因可能是因为线程不一致
    		bpmEngineConfiguration = context.getBean("processEngineConfiguration",BpmEngineConfiguration.class);
    		//BpmServiceUtils.getBpmEngineConfiguration() 采用这个方式获取也可以
    		this.start();
    	}
    }

[SouthEast]: /images/20220713/cff6f0b8d1dd435cae31aaffe91a2730.png