RabbitMQ實現消息的發送和數據同步

引言

    最近參與了一個智慧城市綜合管理平台的項目，主要核心業務就是針對視頻監控管理統計分析城市車輛的相關信息，涉及到幾個平台以及和其他公司合作開發業務的場景，需要對車輛數據進行同步和共享，中間就用到了RabbitMQ的消息中間件，對車輛的大數據進行發送監控和不同系統間的信息同步，下邊就簡單梳理講解一下RabbitMQ發送消息的過程和業務。具體的RabbitMQ服務端和客戶端的配置再此就不講解了

  1：車輛大數據的 BMS同步車輛大數據消息實體CLS_VO_Ignite_Message 

package com.tiandy.easy7.core.vo;

import java.util.List;

/**
 * 同步車輛大數據消息結構
 */
public class CLS_VO_Ignite_Message {

    private String source;                          //標識消息來源於BMS系統
    private String type;                            //user-用戶
    private String operate;                         //add-新增，del-刪除，update-更新
    private CLS_VO_Ignite_User user;                //用戶信息
    private List<CLS_VO_Ignite_Role> role_list;     //用戶的權限信息
    private List<CLS_VO_Ignite_Tollgate> tollgate_list;                 //卡口對應角色清單
    private List<CLS_VO_Ignite_Camera> camera_list;                   //相機對應角色清單
    private List<CLS_VO_AreaParam> area_list;
    private List<CLS_VO_GisCoordinates> gis_list;   //經緯度集合
    private CLS_VO_DomainInfoEx domainInfo;
    public CLS_VO_Ignite_Message(){}

    public CLS_VO_Ignite_Message(String source, String type, String operate, CLS_VO_Ignite_User user, List<CLS_VO_Ignite_Role> role_list, List<CLS_VO_Ignite_Tollgate> tollgate_list, List<CLS_VO_Ignite_Camera> camera_list,List<CLS_VO_AreaParam> area_list) {
        this.source = source;
        this.type = type;
        this.operate = operate;
        this.user = user;
        this.role_list = role_list;
        this.tollgate_list = tollgate_list;
        this.camera_list = camera_list;
        this.area_list = area_list;
    }

    public CLS_VO_Ignite_Message(String source,String type, String operate, List<CLS_VO_GisCoordinates> gis_list) {
        this.source = source;
        this.type = type;
        this.operate = operate;
        this.gis_list = gis_list;
    }
    
    public CLS_VO_Ignite_Message(String source,String type, String operate, CLS_VO_DomainInfoEx domainInfo) {
        this.source = source;
        this.type = type;
        this.operate = operate;
        this.domainInfo = domainInfo;
    }

    public CLS_VO_DomainInfoEx getDomainInfo() {
        return domainInfo;
    }

    public void setDomainInfo(CLS_VO_DomainInfoEx domainInfo) {
        this.domainInfo = domainInfo;
    }

    public String getSource() {
        return source;
    }

    public void setSource(String source) {
        this.source = source;
    }

    public String getType() {
        return type;
    }

    public void setType(String type) {
        this.type = type;
    }

    public String getOperate() {
        return operate;
    }

    public void setOperate(String operate) {
        this.operate = operate;
    }

    public CLS_VO_Ignite_User getUser() {
        return user;
    }

    public void setUser(CLS_VO_Ignite_User user) {
        this.user = user;
    }

    public List<CLS_VO_Ignite_Role> getRole_list() {
        return role_list;
    }

    public void setRole_list(List<CLS_VO_Ignite_Role> role_list) {
        this.role_list = role_list;
    }

    public List<CLS_VO_Ignite_Tollgate> getTollgate_list() {
        return tollgate_list;
    }

    public void setTollgate_list(List<CLS_VO_Ignite_Tollgate> tollgate_list) {
        this.tollgate_list = tollgate_list;
    }

    public List<CLS_VO_Ignite_Camera> getCamera_list() {
        return camera_list;
    }

    public void setCamera_list(List<CLS_VO_Ignite_Camera> camera_list) {
        this.camera_list = camera_list;
    }

    public List<CLS_VO_AreaParam> getArea_list() {
        return area_list;
    }

    public void setArea_list(List<CLS_VO_AreaParam> area_list) {
        this.area_list = area_list;
    }

    public List<CLS_VO_GisCoordinates> getGis_list() {
        return gis_list;
    }

    public void setGis_list(List<CLS_VO_GisCoordinates> gis_list) {
        this.gis_list = gis_list;
    }
}

2：發送RabbitMQ消息的幫助類RabbitMQClientSend

package com.tiandy.easy7.its.rabbitmq;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import cn.jpush.api.utils.StringUtils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.tiandy.easy7.core.bo.CLS_BO_User;
import easy7.datatype.CLS_Easy7_Error;
import easy7.datatype.CLS_Easy7_Types;
import org.apache.log4j.Logger;

public class CLS_RabbitMQClientSend {

    private static final Logger log = Logger.getLogger(CLS_RabbitMQClientSend.class);

    //rabbitmq連接
    private static Connection connection = null;
    //rabbitmq通道
    private static Channel channel = null ;
    //連接狀態標識
    public static boolean connectStatus = false;

    public Connection getConnection() {
        return connection;
    }
    public Channel getChannel() {
        return channel;
    }
    //初始化rabbitmq連接工廠和通道
    public static void initialize(){
        try {
            //連接工廠
            ConnectionFactory factory= CLS_RabbitMQUtil.getRabbitMQConnectionFactory();
            //關閉連接與通道
            closeConnection();
            connection = factory.newConnection();
            channel = connection.createChannel();

            try {
                //聲明一個持久化的交換器，名稱為BMS_SYNC_EXCHANGE_NAME 類型為FANOUT（廣播模式）
                channel.exchangeDeclare("bms_sync_user_auth", BuiltinExchangeType.FANOUT ,true);
                connectStatus = true ;
            } catch (Exception e) {
                connectStatus = false;
                e.printStackTrace();
            }
        } catch (Exception e) {
            connectStatus = false ;
            log.error("CLS_RabbitMQClientSend method initialize error!");
        }
    }

    /**
     * 向消息中間件發送消息（從BMS同步到車輛大數據）
     * @param info 消息
     * @return
     */
    public static int sendMsg(String info) {
        //校驗
        if(StringUtils.isEmpty(info) ){
            return CLS_Easy7_Error.ERROR_PARAM;
        }
        try {
            log.debug("CLS_RabbitMQClientSend sendMsg:" + info.getBytes());
            channel.basicPublish("bms_sync_user_auth", "", null, info.getBytes());
            return CLS_Easy7_Error.ERROR_OK;
        } catch (Exception e) {
            log.error("CLS_RabbitMQClientSend method sendMsg error!");
            return CLS_Easy7_Error.ERROR_REQUEST_FAILED;
        }
    }

    //關閉連接
    public static void closeConnection(){
        try {
            if (channel != null) {
                if(channel.isOpen()) {
                    channel.close();
                    channel = null;
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        try {
            if (connection != null) {
                if(connection.isOpen()) {
                    connection.close();
                    connection = null;
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}

3：RabbitMQ的核心幫助類RabbitMQUtil

package com.tiandy.easy7.its.rabbitmq;

import javax.annotation.Resource;

import com.tiandy.easy7.core.util.Tools;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;

import net.sf.json.JSONObject;

import com.rabbitmq.client.ConnectionFactory;
import com.tiandy.easy7.core.bo.CLS_BO_SystemInfo;
import com.tiandy.easy7.core.po.TabSystemInfo;
import com.tiandy.easy7.core.util.MyApplicationContextUtil;
import com.tiandy.easy7.core.vo.CLS_VO_Result;
import com.tiandy.easy7.face.rabbitmq.CLS_RabbitMQExchangeRecive;

import easy7.datatype.CLS_Easy7_Types;

import java.net.InetAddress;

//MQ工具類
public class CLS_RabbitMQUtil {

    private static final Logger log = LoggerFactory.getLogger(CLS_RabbitMQUtil.class);
    @Resource(name="boSystemInfo")
    private static CLS_BO_SystemInfo boSystemInfo;  
    //構造方法私有
    private CLS_RabbitMQUtil(){
        
    }
    //得到rabbitmq連接工廠
    public static final ConnectionFactory getRabbitMQConnectionFactory(){
        //平台重啟從數據庫獲取rabbitMq的IP
        getRabbitMQIP();
        //當平台第一次使用時，默認為127.0.0.1
        if(null == CLS_Easy7_Types.rabbitmq_host_for_bj || "".equals(CLS_Easy7_Types.rabbitmq_host_for_bj)){
            CLS_Easy7_Types.rabbitmq_host_for_bj = "127.0.0.1";
        }
        RabbitMQConnectionFactory.factory.setHost(CLS_Easy7_Types.rabbitmq_host_for_bj); //127.0.0.1
        return RabbitMQConnectionFactory.factory;
    }
    
    /**
     * 獲取RabbitMQ IP地址
     */
    public static void getRabbitMQIP() {
        ApplicationContext ctx = MyApplicationContextUtil.getContext();
        CLS_BO_SystemInfo boSystemInfo = ctx.getBean(CLS_BO_SystemInfo.class);
        CLS_VO_Result result = null;
        JSONObject json = null;
        JSONObject jsonContent = null;
        JSONObject jsonParam = null;
        int ret;
        String rabbitmq_ip = null;
        try {
            //LS_Easy7_Types.EASY7_ACTIVEMQ_SID ="activemq-server-config-2017728";//activemqid
            result = boSystemInfo.getSystemInfo(CLS_Easy7_Types.EASY7_ACTIVEMQ_SID);
            json = JSONObject.fromObject(result);
            ret = json.getInt("ret");
            if(ret == 0) {
                String content = json.getString("content");
                if(content != null && !"".equals(content)) {
                    jsonContent = JSONObject.fromObject(content);
                    String sParam = jsonContent.getString("sParam");
                    if(sParam != null && !"".equals(sParam)) {
                        jsonParam = JSONObject.fromObject(sParam);
                        rabbitmq_ip = jsonParam.getString("rabbitmqIp");
                    }
                }
            }
        } catch (Exception e1) {
            e1.printStackTrace();
            log.error("rabbitmq id is not exist!");
            return;
        }
        if(jsonParam != null) {
            jsonParam.clear();
            jsonParam = null;
        }
        if(jsonContent != null) {
            jsonContent.clear();
            jsonContent = null;
        }
        if(json != null) {
            json.clear();
            json = null;
        }
        
        if(result == null || rabbitmq_ip == null || "".equals(rabbitmq_ip)) {
            log.error("rabbitmq ip is not exist!");
            return;
        }
          "127.0.0.1" = rabbitmq_ip;

        //獲取本地服務器IP，作為接受下級權限申請的routinfKey
        try{
            if(null == CLS_Easy7_Types.LOCALHOST_IP || "".equals(CLS_Easy7_Types.LOCALHOST_IP)){
                CLS_Easy7_Types.LOCALHOST_IP = Tools.getLinuxLocalIp();
            }
        }catch (Exception e){
            log.error("get LinuxIp error" + e);
        }
    }
    
    
    //rabbitmq連接工廠單例
    private static class RabbitMQConnectionFactory{
        private static ConnectionFactory factory = new ConnectionFactory(); 
        static{
            try {
               //可以通過properties配置文件配置
                factory.setPort(5673);//MQ端口  
                factory.setUsername(admin);//MQ用戶名  
                factory.setPassword(123456);//MQ密碼  
                factory.setRequestedHeartbeat(10);//設置心跳 (秒)
                factory.setAutomaticRecoveryEnabled(true);//自動恢復連接
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

4：獲取需要發送MQ的消息

public CLS_VO_Ignite_Message deleteHostCamera(){
    //定義一個空的list模擬數據，具體實際業務中里邊肯定是有數據的(需要發現車輛的數據消息)
    List<CLS_VO_Ignite_Camera> cameralist = new ArrayList<CLS_VO_Ignite_Camera>();
    //定義拼接向Rabbit發送的消息體 (json格式的數據)
    CLS_VO_Ignite_Message message = new CLS_VO_Ignite_Message(CLS_Easy7_Types.MESSAGE_SOURCE_BMS,CLS_Easy7_Types.MESSAGE_TYPE_CAMERA,
            CLS_Easy7_Types.MESSAGE_OPERATE_DEL,null,null,null,cameralist,null);
    return message;
}

5：發送消息

public void sendRabbitMQMessage(){
  //如發送的消息：
  //{"id":"f6560157-24ae-475b-9736-34072684b008","reviewInfo":"","status":1,"ids":[],"currentUserId":"admin","reviewUser":"admin"}
  CLS_VO_Ignite_Message message =deleteHostCamera();
  
  //消息為空時不發送消息
  if(null != message.getSource() && !"".equals(message.getSource())){
    int sendResult = CLS_RabbitMQClientSend.sendMsg(JSONObject.fromObject(message).toString());
    //同步消息失敗，錯誤代碼使用-40，數據庫操作回滾
    if(sendResult != 0){
        TransactionAspectSupport.currentTransactionStatus().setRollbackOnly(); //事務回滾
        result.setRet( -40);//請求失敗
        return result;
    }
  }
}

6：接收MQ中發送的消息

package com.tiandy.vbs.ics.rabbitMQ;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import javax.annotation.PostConstruct;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.tiandy.vbs.adapter.rabbitmq.CLS_ConsumerThreadPool;
import com.tiandy.vbs.adapter.rabbitmq.CLS_RabbitMQQueueRecive;
import com.tiandy.vbs.adapter.rabbitmq.CLS_RabbitMQUtil;
import com.tiandy.vbs.common.util.CLS_VBS_Types;
import com.tiandy.vbs.common.util.InterfaceLogger;
import com.tiandy.vbs.ics.bo.CLS_BO_DispositionAdapter;
import com.tiandy.vbs.ics.vo.CLS_VO_MQMessage;

@Component
public class CLS_QueueDispositionRecive extends CLS_RabbitMQQueueRecive{
    //rabbitmq連接
    private static Connection connection = null;
    //rabbitmq通道
    private static Channel channel = null ;
    //連接狀態標識
    public boolean connectStatus = false;
    //消費者線程池
    public static CLS_ConsumerThreadPool consumerThreadPool = new CLS_ConsumerThreadPool(CLS_VBS_Types.corePoolSize,CLS_VBS_Types.maximumPoolSize,
            CLS_VBS_Types.keepAliveTime, TimeUnit.SECONDS,new ArrayBlockingQueue<Runnable>(CLS_VBS_Types.workQueue), new ThreadPoolExecutor.CallerRunsPolicy());
    @Autowired
    private CLS_BO_DispositionAdapter boAdapter;
    public static CLS_QueueDispositionRecive queueRecive;    
    @PostConstruct
    public void init() {
        queueRecive = this;
    }
    public Connection getConnection() {
        return connection;
    }
    public Channel getChannel() {
        return channel;
    }
    public void initialize(){
        try {
            //連接工廠
            ConnectionFactory factory= CLS_RabbitMQUtil.getRabbitMQConnectionFactory();
            //關閉連接與通道
            closeConnection();
            connection = factory.newConnection();  
            channel = connection.createChannel();  
            syncConsumer();
            connectStatus = true ;
        } catch (Exception e) {
            connectStatus = false ;
            e.printStackTrace();
            InterfaceLogger.error("CLS_RabbitMQQueueRecive method initialize:"+e.getMessage(),e);
        }
    }
       //核心同步接收消息
    public void syncConsumer()throws Exception{
        Map<String, Object> args = new HashMap<String, Object>();
        args.put("x-max-length", 100000);
        args.put("x-message-ttl",CLS_VBS_Types.x_message_ttl);

        channel.queueDeclare(CLS_VBS_Types.rabbitmq_queue_alarm_record,true,false,false,args);
        //聲明交換器 
        channel.exchangeDeclare(CLS_VBS_Types.rabbitmq_exchange_name, BuiltinExchangeType.FANOUT ,true);
        channel.queueBind(CLS_VBS_Types.rabbitmq_queue_alarm_record, CLS_VBS_Types.rabbitmq_exchange_name, "");  
        //消費者
        Consumer consumer = new DefaultConsumer(channel) {  
            @Override  
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                //接收到的消息
                String message = new String(body, "UTF-8");
                InterfaceLogger.debug("recive message:"+message);
                if(!"".equals(message)){                    
                        CLS_VO_MQMessage mqVo = new CLS_VO_MQMessage();
                        mqVo.setMessage(message);
                        mqVo.setEnvelope(envelope);
                        CLS_DispositionConsumer worker = new CLS_DispositionConsumer(mqVo,queueRecive.boAdapter,channel);
                        consumerThreadPool.submit(worker);//多線程處理數據
                }else {
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            }  
        };  
        //消息手動確認
        channel.basicConsume(CLS_VBS_Types.rabbitmq_queue_alarm_record, false, consumer); 
    }    
    //關閉連接
    public void closeConnection(){
           try {
               if (channel != null) {
                  if(channel.isOpen()) {
                      channel.close();
                      channel = null;
                   }
                 }
            } catch (Exception e) {
                InterfaceLogger.error("CLS_RabbitMQExchangeRecive closeChannel  error  " + e);
                e.printStackTrace();
            }
            try {
                if (connection != null) {
                  if(connection.isOpen()) {
                      connection.close();
                      connection = null;
                     }
                }
           } catch (Exception e) {
                InterfaceLogger.error("CLS_RabbitMQExchangeRecive closeConnection  error  " + e);
                e.printStackTrace();
             }
        }
}

7：線程池處理消息類

package com.tiandy.vbs.adapter.rabbitmq;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.springframework.stereotype.Component;
import com.tiandy.vbs.common.util.CLS_VBS_Types;
@Component
public abstract class CLS_RabbitMQQueueRecive {
    
    //消費者線程池
    public static CLS_ConsumerThreadPool consumerThreadPool = new CLS_ConsumerThreadPool(CLS_VBS_Types.corePoolSize,CLS_VBS_Types.maximumPoolSize,
            CLS_VBS_Types.keepAliveTime, TimeUnit.SECONDS,new ArrayBlockingQueue<Runnable>(CLS_VBS_Types.workQueue), new ThreadPoolExecutor.CallerRunsPolicy());    
    //初始化連接
    public abstract void initialize();
    //消費者
    public abstract void syncConsumer()throws Exception;
    //關閉連接
    public abstract void closeConnection();
}

其他應用或平台接收到MQ中的消息后，解析JSON數據，轉換為對於的數據實體，將數據添加到相應的庫中表中即可，這樣就完成了消息的發送和數據的同步！

至此，發現消息得過程就完成了！