您现在的位置是：首页 > Java

当前栏目

java集成ActiveMq

JAVA 消息代码

2023-03-02 11:14:37 时间

#### 1.安装activeMq ####

我这里是在linux中安装的，[博客][Link 1]中只记录了单机安装，安装集群，可以自己搜索下。

#### 2.引入active的相关jar ####

maven中的代码

<dependency>
      <groupId>org.apache.activemq</groupId>
      <artifactId>activemq-broker</artifactId>
      <version>5.14.5</version>
    </dependency>
    <dependency>
      <groupId>org.apache.activemq</groupId>
      <artifactId>activemq-pool</artifactId>
      <version>5.14.5</version>
    </dependency>
    <dependency>
      <groupId>org.apache.activemq</groupId>
      <artifactId>activemq-spring</artifactId>
      <version>5.14.5</version>
    </dependency>
    <dependency>
      <groupId>org.apache.activemq</groupId>
      <artifactId>activemq-client</artifactId>
      <version>5.14.5</version>
    </dependency>

#### 3.配置mq相关信息 ####

**mq.properties中代码**

## 这里是集群，可以配置单个的
    mq.brokerURL=failover:(tcp://192.168.1.94:51511,tcp://192.168.1.94:51512,
    tcp://192.168.1.94:51513)?Randomize=false
    mq.userName=admin
    mq.password=admin

**mqContext.xml中代码**

<?xml version="1.0" encoding="UTF-8"?>
    <beans xmlns="http://www.springframework.org/schema/beans"
    	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:aop="http://www.springframework.org/schema/aop"
    	xmlns:tx="http://www.springframework.org/schema/tx" xmlns:context="http://www.springframework.org/schema/context"
    	xmlns:jee="http://www.springframework.org/schema/jee" xmlns:amq="http://activemq.org/config/1.0"
    	xsi:schemaLocation=" 
              http://www.springframework.org/schema/jee 
              http://www.springframework.org/schema/jee/spring-jee-4.0.xsd
              http://www.springframework.org/schema/beans
              http://www.springframework.org/schema/beans/spring-beans-4.0.xsd 
              http://www.springframework.org/schema/tx 
              http://www.springframework.org/schema/tx/spring-tx-4.0.xsd
              http://www.springframework.org/schema/context 
              http://www.springframework.org/schema/context/spring-context-4.0.xsd 
              http://www.springframework.org/schema/aop 
              http://www.springframework.org/schema/aop/spring-aop-4.0.xsd"
    	default-autowire="byName">
    	
    	
    	<bean id="connectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop">
    		<property name="connectionFactory">
    			<bean class="org.apache.activemq.ActiveMQConnectionFactory">
    				<property name="brokerURL" value="${mq.brokerURL}" />
    				<property name="userName" value="${mq.userName}"/>
    				<property name="password" value="${mq.password}"/>
    				
    			</bean>
    		</property>
    		<property name="maxConnections" value="10"></property>
    	</bean>
         
    	<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
    		<property name="connectionFactory" ref="connectionFactory" />
    		<property name="deliveryMode" value="1" />
    	</bean>
    	 
    	 <bean id="MessageSender" class="com.common.mq.dao.MqSender">  
    		<constructor-arg name="jmsTemplate" ref="jmsTemplate"/>
    	</bean> 
    	</beans>

配置好后，需要在spring的xml中将mqContent.xml注入并配置参数。

#### 4.生产者部分代码 ####

import javax.annotation.Resource;
    import javax.jms.Connection;
    import javax.jms.DeliveryMode;
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.MessageProducer;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    import org.springframework.jms.core.JmsTemplate;
    
    public class MqSender {
    	private JmsTemplate jmsTemplate;
    	private Connection connection=null;
    	private Session session;
    	private Destination destination;
    	private MessageProducer producer;
    	@Resource
    	private LogService logService;
    	
    	/*构造MqSender时将jmsTemplate注入*/
    	public MqSender(JmsTemplate jmsTemplate){
    		this.jmsTemplate=jmsTemplate;
    		try {
    			this.connection=this.jmsTemplate.getConnectionFactory().createConnection();
    			connection.start();
    			this.session=connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
    			this.producer = session.createProducer(null);
    		}catch(JMSException e){
    			logService.saveLog(e.toString(), this.getClass(), 4);
    			e.printStackTrace();
    		}
    	}
    	
    	public void sendMsg(String msgJson,String queueName){
    		TextMessage omsg;
    		try {
    			omsg=session.createTextMessage(msgJson);
    			// 第一个参数目标地址
    			// 第二个参数 具体的数据信息
    			// 第三个参数 传送数据的模式
    			// 第四个参数 优先级
    			// 第五个参数 消息的过期时间
    			this.destination = session.createQueue(queueName);
    			producer.send( this.destination,omsg, DeliveryMode.NON_PERSISTENT, 0 , 0);
    		} catch (JMSException e) {
    			logService.saveLog(e.toString(), this.getClass(), 4);
    			e.printStackTrace();
    		}
    		omsg=null;
    	}
    	
    	public JmsTemplate getJmsTemplate() {
    		return jmsTemplate;
    	}
    	public void setJmsTemplate(JmsTemplate jmsTemplate) {
    		this.jmsTemplate = jmsTemplate;
    	}
    }

**iMessageService部分代码**

public interface IMessageService {
    	public String MSG_TYPE_IDENTIFY="LOCAL_IDENTIFY";
    	public String MSG_TYPE_USERSCORE="LOCAL_USERSCORE";
    	public String MSG_TYPE_ORDER="LOCAL_ORDER";
    	public String MSG_TYPE_SYSINFO="LOCAL_SYSINFO";
    	public String MSG_TYPE_MSG_NOTICE="LOCAL_NOTICE";
    	public String MSG_TYPE_COMMENT="LOCAL_COMMENT";
    	/*public String MSG_TYPE_IDENTIFY="IDENTIFY";
    	public String MSG_TYPE_USERSCORE="USERSCORE";
    	public String MSG_TYPE_ORDER="ORDER";
    	public String MSG_TYPE_SYSINFO="SYSINFO";
    	public String MSG_TYPE_MSG_NOTICE="NOTICE";
    	public String MSG_TYPE_COMMENT="COMMENT";*/
    	public void sendIdentifyMsg(IMsgContent msg);
    	public void sendUserscoreMsg(IMsgContent msg);
    	public void sendOrderMsg(IMsgContent msg);
    	public void sendSysinfoMsg(IMsgContent msg);
    	public void sendMsgNotice(IMsgContent msg);
    	public void sendComment(IMsgContent msg);
    }

**service实现类**

import javax.annotation.Resource;
    import org.springframework.stereotype.Service;
    import com.alibaba.fastjson.JSONObject;
    
    @Service("activeMessageService")
    public class MessageServiceImpl implements IMessageService{
    	@Resource
    	private MqSender MessageSender;
    	@Resource
    	private LogService logService;
    	@Override
    	public void sendIdentifyMsg(IMsgContent msg) {
    		commonMsgHandler(msg, MSG_TYPE_IDENTIFY);
    	}
    
    	@Override
    	public void sendUserscoreMsg(IMsgContent msg) {
    		commonMsgHandler(msg, MSG_TYPE_USERSCORE);
    	}
    
    	@Override
    	public void sendOrderMsg(IMsgContent msg) {
    		commonMsgHandler(msg, MSG_TYPE_ORDER);
    	}
    
    	@Override
    	public void sendSysinfoMsg(IMsgContent msg) {
    		commonMsgHandler(msg, MSG_TYPE_SYSINFO);
    	}
    	
    	@Override
    	public void sendMsgNotice(IMsgContent msg) {
    		commonMsgHandler(msg, MSG_TYPE_MSG_NOTICE);
    	}
    	@Override
    	public void sendComment(IMsgContent msg) {
    		commonMsgHandler(msg, MSG_TYPE_COMMENT);
    	}
    	public void commonMsgHandler(IMsgContent msg,String msgType)
    	{
    		try {
    			MessageSender.sendMsg(JSONObject.toJSONString(msg),msgType);
    		} catch (Exception e) {
    			logService.saveLog(e.toString(), this.getClass(), 4);
    			e.printStackTrace();
    		}
    	}
    }

**具体的发送消息是通过集成ImsgContent的类进行封装的。在调用时候只需要使用iMessageService中每个方法，具体传递的参数需要new 具体的对象。对象中只需要定义参数并提供set和get方法并将参数传入构造方法中。**

#### 5.接收消息代码 ####

接收部分需要新建一个项目进行运行，专门处理mq队列中的消息。其参数注入和配置和之前一样，区别在于接收消息部分代码。具体mqReceive代码如下：

import java.util.List;
    import javax.annotation.Resource;
    import javax.jms.Connection;
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageConsumer;
    import javax.jms.MessageListener;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    import com.google.gson.JsonParser;
    import org.springframework.jms.core.JmsTemplate;
    import com.google.gson.JsonObject;
    
    public class MqReicever{
    	private  Connection connection=null;
    	private  Session session;
    	private  Destination destinationIDENTIFY;
    	private  Destination destinationUSERSCORE;
    	private  Destination destinationORDER;
    	private  Destination destinationSYSINFO;
    	private  Destination destinationMSGNOTICE;
    	private  Destination destinationCOMMENT;
    	private  MessageConsumer mConsumerIdentify;
    	private  MessageConsumer mConsumerUserscore;
    	private  MessageConsumer mConsumerOrder;
    	private  MessageConsumer mConsumerSysinfo;
    	private  MessageConsumer mConsumerMsgNotice;
    	private  MessageConsumer mConsumerComment;
    	private JmsTemplate jmsTemplate;
    	@Resource(name="memberService")
    	private MemberService memberService;
        @Resource
        private LogService logService;
    	@Resource(name ="sendMsg") 
    	private SendMsg sendMsg;
    	//注册
    	public MqReicever(JmsTemplate jmsTemplate){
    		this.jmsTemplate=jmsTemplate;
    		try {
    			this.connection=this.jmsTemplate.getConnectionFactory().createConnection();
    			connection.start();
    			this.session=connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
    			
    			this.destinationIDENTIFY = session.createQueue(IMessageService.MSG_TYPE_IDENTIFY);
    			this.destinationUSERSCORE = session.createQueue(IMessageService.MSG_TYPE_USERSCORE);
    			this.destinationORDER = session.createQueue(IMessageService.MSG_TYPE_ORDER);
    			this.destinationSYSINFO = session.createQueue(IMessageService.MSG_TYPE_SYSINFO);
    			this.destinationMSGNOTICE = session.createQueue(IMessageService.MSG_TYPE_MSG_NOTICE);
    			this.destinationCOMMENT = session.createQueue(IMessageService.MSG_TYPE_COMMENT);
    			this.mConsumerIdentify = session.createConsumer(destinationIDENTIFY);
    			this.mConsumerUserscore = session.createConsumer(destinationUSERSCORE);
    			this.mConsumerOrder = session.createConsumer(destinationORDER);
    			this.mConsumerSysinfo = session.createConsumer(destinationSYSINFO);
    			this.mConsumerMsgNotice = session.createConsumer(destinationMSGNOTICE);
    			this.mConsumerComment = session.createConsumer(destinationCOMMENT);
    			this.mConsumerIdentify.setMessageListener(new MessListenerIdentify());
    			this.mConsumerUserscore.setMessageListener(new MessListenerUserscore());
    			this.mConsumerOrder.setMessageListener(new MessListenerOrder());
    			this.mConsumerSysinfo.setMessageListener(new MessListenerSysinfo());
    			this.mConsumerMsgNotice.setMessageListener(new MessListenerMsgNotice());
    			this.mConsumerComment.setMessageListener(new MessListenerComment());
    		} catch (JMSException e) {
    			logService.saveLog(e.toString(), this.getClass(), 4);
    			e.printStackTrace();
    		}
    	}
    	
    	//处理msgNotice消息
    	class MessListenerMsgNotice implements MessageListener{
    		@Override
    		public void onMessage(Message message) {
    			if(message instanceof TextMessage){
    				TextMessage om=(TextMessage) message;
    				try {
    					String msgNotice= om.getText();
    					JsonParser parser = new JsonParser();
    					JsonObject jsonObject = (JsonObject) parser.parse(msgNotice);
    					String memberName = jsonObject.get("memberName").getAsString();
    					String phone = jsonObject.get("phone").getAsString();
    					String msgType = jsonObject.get("msgType").getAsString();
    					sendMsg.sendMsgNotice(memberName, phone, msgType);
    				} catch (JMSException e) {
    					logService.saveLog(e.toString(), this.getClass(), 4);
    					e.printStackTrace();
    				}
    			}
    		}
    	}
    	
    	//处理评论内容
    	class MessListenerComment implements MessageListener{
    		@Override
    		public void onMessage(Message message) {
    			if(message instanceof TextMessage){
    				TextMessage om=(TextMessage) message;
    				try {
    					String msgNotice= om.getText();
    					JsonParser parser = new JsonParser();
    					JsonObject jsonObject = (JsonObject) parser.parse(msgNotice);
    					System.out.println(jsonObject);
    				} catch (JMSException e) {
    					logService.saveLog(e.toString(), this.getClass(), 4);
    					e.printStackTrace();
    				}
    			}
    		}
    	}
    	
    	//处理identify消息
    	class MessListenerIdentify implements MessageListener{
    		@Override
    		public void onMessage(Message message) {
    			if(message instanceof TextMessage){
    				TextMessage om=(TextMessage) message;
    				try {
    					String identifyMsg= om.getText();
    					JsonParser parser = new JsonParser();
    					JsonObject jsonObject = (JsonObject) parser.parse(identifyMsg);
    					Integer code = jsonObject.get("code").getAsInt();
    					String phone = jsonObject.get("phone").getAsString();
    					String identityType = jsonObject.get("identityType").getAsString();
    					sendMsg.sendIdentity(phone, code, identityType);
    				} catch (JMSException e) {
    					logService.saveLog(e.toString(), this.getClass(), 4);
    					e.printStackTrace();
    				}
    			}
    		}
    	}
    	
    	//处理userscore 消息
    	class MessListenerUserscore implements MessageListener{
    		@Override
    		public void onMessage(Message message) {
    			if(message instanceof TextMessage){
    				TextMessage om=(TextMessage) message;
    				try {
    					String userScoreMsg= om.getText();
    					JsonParser parser = new JsonParser();
    					JsonObject jsonObject = (JsonObject) parser.parse(userScoreMsg);
    					Integer memberId = jsonObject.get("memberId").getAsInt();
    					Integer number = jsonObject.get("number").getAsInt();
    					Integer updateType = jsonObject.get("updateType").getAsInt();
    					String reason = jsonObject.get("reason").getAsString();
    					sendMsg.integralChange(memberId, number, updateType, reason);
    				} catch (JMSException e) {
    					logService.saveLog(e.toString(), this.getClass(), 4);
    					e.printStackTrace();
    				}
    			}
    		}
    	}
    
    	//处理 sysinfo 消息
    	class MessListenerSysinfo implements MessageListener{
    		@Override
    		public void onMessage(Message message) {
    			if(message instanceof TextMessage){
    				TextMessage om=(TextMessage) message;
    				try {
    					String sysInfoMsg= om.getText();
    					JsonParser parser = new JsonParser();
    					JsonObject jsonObject = (JsonObject) parser.parse(sysInfoMsg);
    					String receiverType = jsonObject.get("receiverType").getAsString();
    					String messageContent = jsonObject.get("messageContent").getAsString();
    					String msgTitle = jsonObject.get("msgTitle").getAsString();
    					List<Member> memberList = memberService.findList();
    					sendMsg.sendSystemMsg(receiverType, messageContent, msgTitle, memberList);
    				} catch (JMSException e) {
    					logService.saveLog(e.toString(), this.getClass(), 4);
    					e.printStackTrace();
    				}
    			}
    		}
    	}
    );
    
    public JmsTemplate getJmsTemplate() {
    	return jmsTemplate;
    }
    
    public void setJmsTemplate(JmsTemplate jmsTemplate) {
    		this.jmsTemplate = jmsTemplate;
    	}
    }

[Link 1]: https://blog.csdn.net/github_38924695/article/details/78649296

猜你喜欢

Python入门（04） -- 函数
Amazon Braket – 开始使用量子计算
Java并发——Semaphore
Python虚拟机实现
Java发工资
AWS DeepRacer 更新 – 新功能和新的竞赛机会
python+selenium面试题
java中RuntimeException
Java初始化顺序
python多线程
AWS Single Sign-On 的新功能
Java并发包
python中文件操作
java CAS操作
新增功能 – AWS IoT Greengrass 在边缘增加了容器支持和数据流管理功能
Python 绝对简明手册
python基础：os模块
Java Unsafe类
欢迎来到 AWS IoT 日 – 8 项强大的新功能
Python学习-基础（四）

zl程序教程

当前栏目

java集成ActiveMq

相关文章