欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

activeMq集成 博客分类: activeMq  

程序员文章站 2024-03-19 19:53:22
...

activeMq集成:

mq类似redis安装好了软件之后,就在java端集成,用工厂工具类调用,mq用于消息推送(结合页面的轮询做聊天工具),结合redis缓解高并发压力---来不及处理的放在队列中,

多系统的事务回滚通知

 

pom.xml

<!-- activemq相关依赖 -->

<dependency>

<groupId>javax.jms</groupId>

<artifactId>jms</artifactId>

<version>1.1</version>

</dependency>

<!-- <dependency> <groupId>com.sun.jmx</groupId> <artifactId>jmxri</artifactId> 

<version>1.2.1</version> </dependency> <dependency> <groupId>com.sun.jdmk</groupId> 

<artifactId>jmxtools</artifactId> <version>1.2.1</version> </dependency> -->

 

<dependency>

<groupId>org.apache.activemq</groupId>

<artifactId>activemq-core</artifactId>

<version>5.1.0</version>

</dependency>

<dependency>

<groupId>org.springframework</groupId>

<artifactId>spring-jms</artifactId>

<version>3.2.16.RELEASE</version>

</dependency>

<dependency>

<groupId>net.sf.json-lib</groupId>

<artifactId>json-lib</artifactId>

<classifier>jdk15</classifier>

<version>2.4</version>

</dependency>

<!-- activemq相关依赖 Over -->

 

application.xml

 <!-- 引入activemq -->

    <import resource="activemq.xml" />

 

activemq.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:p="http://www.springframework.org/schema/p"

xmlns:context="http://www.springframework.org/schema/context"

xmlns:jms="http://www.springframework.org/schema/jms" xmlns:amq="http://activemq.apache.org/schema/core"

xmlns:aop="http://www.springframework.org/schema/aop"

xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.0.xsd http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-3.0.xsd http://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms-3.0.xsd http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core-5.5.0.xsd">

 

<bean id="topicSendConnectionFactory" class="org.apache.activemq.spring.ActiveMQConnectionFactory">

<!-- <property name="brokerURL" value="udp://localhost:8123" /> -->

<!-- UDP传输方式 -->

<property name="brokerURL" value="tcp://10.0.1.126:61616" />

<!-- TCP传输方式 -->

<property name="useAsyncSend" value="true" />

</bean>

 

<bean id="topicListenConnectionFactory" class="org.apache.activemq.spring.ActiveMQConnectionFactory">

<!-- <property name="brokerURL" value="udp://localhost:8123" /> -->

<!-- UDP传输方式需要在activemq上面做配置 -->

<property name="brokerURL" value="tcp://10.0.1.126:61616" />

<!-- TCP传输方式 -->

</bean>

<!-- 定义主题 -->

<bean id="myTopic" class="org.apache.activemq.command.ActiveMQTopic">

<constructor-arg value="esteelMessage-mq" />

</bean>

 

<bean id="messageConvertForSys" class="com.esteel.message.mq.MessageConvertForSys" />

 

<!-- TOPIC send jms模板 -->

<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">

<property name="connectionFactory" ref="topicSendConnectionFactory" />

<property name="defaultDestination" ref="myTopic" />

<property name="messageConverter" ref="messageConvertForSys" />

<!-- 发送模式 DeliveryMode.NON_PERSISTENT=1:非持久 ; DeliveryMode.PERSISTENT=2:持久 -->

<property name="deliveryMode" value="1" />

<property name="pubSubDomain" value="true" />

<!-- 开启订阅模式 -->

</bean>

<!-- 消息发送方 -->

<bean id="topicSender" class="com.esteel.message.mq.MessageSender">

<property name="jmsTemplate" ref="jmsTemplate" />

</bean>

 

<!-- <bean id="springContextUtil" class="com.esteel.common.SpringContextUtil" /> -->

    

 

<!-- 消息接收方 -->

<bean id="topicReceiver" class="com.esteel.message.mq.MessageReceiver" />

<!-- 主题消息监听容器,一经注册,自动监听 -->

<bean id="listenerContainer"

class="org.springframework.jms.listener.DefaultMessageListenerContainer">

<property name="connectionFactory" ref="topicListenConnectionFactory" />

<property name="pubSubDomain" value="true" />

<!-- true 订阅模式 -->

<property name="destination" ref="myTopic" />

<!-- 目的地 myTopic -->

<property name="subscriptionDurable" value="true" />

<!-- -这里是设置接收客户端的ID,在持久化时,但这个客户端不在线时,消息就存在数据库里,直到被这个ID的客户端消费掉 -->

<property name="clientId" value="clientId_esteelMessage_1" />

<property name="messageListener" ref="topicReceiver" />

</bean>

<!-- Servlet -->

<!-- <bean id="ControlServlet1" class="com.esteel.servlet.ControlServlet1"> 

<property name="topicSender" ref="topicSender" /> </bean> -->

</beans>  

 

 

自定义发送,接收,转化类:

 

转化:

package com.esteel.message.mq;

 

import org.springframework.jms.core.JmsTemplate;

import org.springframework.stereotype.Component;

 

public class MessageSender {

 

private JmsTemplate jmsTemplate;

 

public void sendMessage(String msg) {

jmsTemplate.convertAndSend(msg);

}

 

public JmsTemplate getJmsTemplate() {

return jmsTemplate;

}

 

public void setJmsTemplate(JmsTemplate jmsTemplate) {

this.jmsTemplate = jmsTemplate;

}

 

}

 

 

 

接收:

 

package com.esteel.message.mq;

 

import java.text.SimpleDateFormat;

import java.util.Calendar;

 

import javax.jms.JMSException;

import javax.jms.Message;

import javax.jms.MessageListener;

import javax.jms.ObjectMessage;

 

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

import org.springframework.stereotype.Controller;

 

import com.esteel.exception.EsteelException;

import com.esteel.message.bean.TbConObj;

import com.esteel.message.bean.TbConOrd;

import com.esteel.message.bean.TbConOrdPrice;

import com.esteel.message.beanVo.TbConOrdVo;

import com.esteel.message.redis.service.RedisService;

import com.esteel.message.service.TbConObjService;

import com.esteel.message.service.TbConOrdPriceService;

import com.esteel.message.service.TbConOrdService;

 

import net.sf.json.JSONObject;

 

@Controller

public class MessageReceiver implements MessageListener {

 

@Autowired

TbConOrdService tbConOrdService;

@Autowired

TbConOrdPriceService tbConOrdPriceService;

@Autowired

TbConObjService tbConObjService;

@Autowired

RedisService redisService;

 

public void onMessage(Message m) {

ObjectMessage om = (ObjectMessage) m;

try {

String key_esteelMessage = om.getStringProperty("key_esteelMessage");

JSONObject object1 = JSONObject.fromObject(key_esteelMessage);

String objectName = (String)object1.get("objectName");

if(objectName.equals("TbConOrdVo")){

JSONObject object2 = (JSONObject) object1.get("object");

TbConOrdVo tbConOrdVo=(TbConOrdVo)JSONObject.toBean(object2, TbConOrdVo.class);

TbConOrd tbConOrd = new TbConOrd();

/*获取ipAddress*/

String ipAddress = (String)object1.get("ipAddress");

/* 从tbConOrdVo中提取tbConOrd */

tbConOrd = copyTbConOrd(tbConOrdVo, tbConOrd, ipAddress);

/* 写入tbConOrd */

tbConOrd = tbConOrdService.insertTbConOrd(tbConOrd);

TbConOrdPrice tbConOrdPrice = new TbConOrdPrice();

tbConOrdPrice = copyTbConOrdPrice(tbConOrd, tbConOrdVo, tbConOrdPrice);

/* 写入聊天文字到 tbConOrdPrice*/

String msgText = tbConOrdPrice.getMsgText();

if (msgText.equals("请录入您的议价留言,最大为300个字符!按Ctrl+Enter提交!")) {

tbConOrdPrice.setMsgText("");

}

//tbConOrd = tbConOrdService.getTbConOrdByObj(tbConOrd);

//tbConOrdPrice.setOrdKey(tbConOrd.getOrdKey());

tbConOrdPrice=tbConOrdPriceService.insertTbConOrdPrice(tbConOrdPrice);

/* 还得写入tbConObj的数据 */

TbConObj tbConObj=tbConObjService.getTbConObjById(tbConOrd.getConobjKey());

if(tbConObj.getContradeType().equals("A")){

/*销售单*/

/*设置最低价位*/

if(tbConObj.getHighPrice()==null){

tbConObj.setHighPrice(tbConObj.getOrderPrice());

}

if(tbConOrdPrice.getOrdpriceMan()==null){

/*当前对象是客人*/

tbConObj.setLowPrice(tbConOrd.getOrderPrice());

}else if(tbConOrdPrice.getOrdpriceMan().equals("A")){

/*当前对象是客人*/

tbConObj.setLowPrice(tbConOrd.getOrderPrice());

}else{

/*当前对象是主人*/

tbConObj.setHighPrice(tbConOrd.getOrderPrice());

}

}else{

/*采购单*/

/*设置最低价位*/

if(tbConObj.getLowPrice()==null){

tbConObj.setLowPrice(tbConObj.getOrderPrice());

}

if(tbConOrdPrice.getOrdpriceMan()==null){

/*当前对象是客人*/

tbConObj.setHighPrice(tbConOrd.getOrderPrice());

}else if(tbConOrdPrice.getOrdpriceMan().equals("A")){

/*当前对象是客人*/

tbConObj.setHighPrice(tbConOrd.getOrderPrice());

}else{

/*当前对象是主人*/

tbConObj.setLowPrice(tbConOrd.getOrderPrice());

}

}

tbConObj = tbConObjService.updateTbConObj(tbConObj);

tbConObjService.listClearCache(tbConObj.getConobjKey());

tbConOrdService.listClearCache();

tbConOrdService.listClearCache(tbConObj.getConobjKey(),tbConOrdVo.getTradeCustomerKey());

}

System.out.println("==============MQ Write to Database success============");

} catch (JMSException e) {

e.printStackTrace();

} catch (EsteelException e) {

e.printStackTrace();

}

 

}

 

private TbConOrd getExistTbConOrd(TbConOrd tbConOrd) throws EsteelException{

TbConOrd tbConOrdNew = new TbConOrd();

tbConOrdNew.setConobjKey(tbConOrd.getConobjKey());

tbConOrdNew.setCustomerKey(tbConOrd.getCustomerKey());

TbConOrd tbConOrdNew2 = new TbConOrd();

tbConOrdNew2 = tbConOrdService.getTbConOrdByObj(tbConOrdNew);

if(null==tbConOrdNew2){

return tbConOrdNew2;

}

return tbConOrdNew2;

}

 

private TbConOrd copyTbConOrd(TbConOrdVo tbConOrdVo, TbConOrd tbConOrd, String ipAddress) {

tbConOrd.setConobjKey(tbConOrdVo.getConobjKey());

tbConOrd.setContradeType(tbConOrdVo.getContradeType());

tbConOrd.setCanTradeTypes(tbConOrdVo.getCanTradeTypes());

tbConOrd.setBailBankCode(tbConOrdVo.getBailBankCode());

tbConOrd.setOrderPrice(tbConOrdVo.getOrderPrice());

tbConOrd.setTradeComm(tbConOrdVo.getTradeComm());

tbConOrd.setDealBail(tbConOrdVo.getDealBail());

tbConOrd.setDisTime(Calendar.getInstance().getTime());

tbConOrd.setDisIp(tbConOrdVo.getDisIp());

tbConOrd.setStartTranDate(tbConOrdVo.getStartTranDate());

tbConOrd.setSubsidyInterest(tbConOrdVo.getSubsidyInterest());

tbConOrd.setBackCause(tbConOrdVo.getBackCause());

tbConOrd.setPickType(tbConOrdVo.getPickType());

tbConOrd.setUponRange(tbConOrdVo.getUponRange());

tbConOrd.setAddsubMark(tbConOrdVo.getAddsubMark());

/* 上面的是从Vo中继承过来的 */

/* 下面的是手动添加的 */

tbConOrd.setTradeDate(new SimpleDateFormat("yyyy-MM-dd").format(Calendar.getInstance().getTime()));

/*用tradeCustomerKey来替换customerKey*/

tbConOrd.setCustomerKey(tbConOrdVo.getTradeCustomerKey());

tbConOrd.setOrderNum(tbConOrdVo.getNewNum());

tbConOrd.setOrderPrice(tbConOrdVo.getNewPrice());

tbConOrd.setCntPrice(tbConOrdVo.getNewPrice());

tbConOrd.setOrderStatus("A");

tbConOrd.setTradeComm("0");

tbConOrd.setDealBail("0");

tbConOrd.setOrderTime(Calendar.getInstance().getTime());

tbConOrd.setOrderIp(ipAddress);

tbConOrd.setDisTime(tbConOrd.getOrderTime());

tbConOrd.setOrderIp(tbConOrd.getOrderIp());

tbConOrd.setSubsidyInterest("0");

tbConOrd.setMarkchkStatus("A");

tbConOrd.setShouldPayBail("0");

tbConOrd.setShouldPayMoney("0");

tbConOrd.setIsBailPayGoods("Y");

tbConOrd.setPickType("A");

return tbConOrd;

}

 

private TbConOrdPrice copyTbConOrdPrice(TbConOrd tbConOrd, TbConOrdVo tbConOrdVo, TbConOrdPrice tbConOrdPrice) throws EsteelException {

tbConOrdPrice.setOrdKey(tbConOrd.getOrdKey());

tbConOrdPrice.setCustomerKey(tbConOrd.getCustomerKey());

tbConOrdPrice.setKfCustomerKey(tbConOrd.getKfCustomerKey());

tbConOrdPrice.setNewNum(tbConOrd.getOrderNum());

tbConOrdPrice.setNewPrice(tbConOrd.getOrderPrice());

tbConOrdPrice.setIsNewprice("Y");

tbConOrdPrice.setTradeComm(tbConOrd.getTradeComm());

tbConOrdPrice.setDealBail(tbConOrd.getDealBail());

tbConOrdPrice.setMsgText(tbConOrdVo.getMsgText());

tbConOrdPrice.setIrdTime(tbConOrd.getOrderTime());

tbConOrdPrice.setOrdpriceType("A");

tbConOrdPrice.setIsLook("N");

tbConOrdPrice.setCdListKeys(tbConOrd.getCdListKeys());

tbConOrdPrice.setStartTranDate(tbConOrd.getStartTranDate());

tbConOrdPrice.setSubsidyInterest(tbConOrd.getSubsidyInterest());

tbConOrdPrice.setUponRange(tbConOrd.getUponRange());

tbConOrdPrice.setAddsubMark(tbConOrd.getAddsubMark());

tbConOrdPrice.setUponRange2(tbConOrd.getUponRange2());

tbConOrdPrice.setAddsubMark2(tbConOrd.getAddsubMark2());

tbConOrdPrice.setLookTimes("0");

tbConOrdPrice.setOrdpriceMan(tbConOrdVo.getOrdpriceMan());

/* 为了取当前的序号,需要获取当前的最新的值,然后+1*/

/* 最后用时间代替了这个number */

tbConOrdPrice.setOrdpriceNo(String.valueOf(System.currentTimeMillis()));

return tbConOrdPrice;

}

 

}

 

 

发送:

package com.esteel.message.mq;

 

import javax.jms.JMSException;

import javax.jms.Message;

import javax.jms.ObjectMessage;

import javax.jms.Session;

 

import org.springframework.jms.support.converter.MessageConversionException;

import org.springframework.jms.support.converter.MessageConverter;

import org.springframework.stereotype.Component;

 

public class MessageConvertForSys implements MessageConverter {

 

public Message toMessage(Object object, Session session) throws JMSException, MessageConversionException {

System.out.println("sendMessage:" + object.toString());

ObjectMessage objectMessage = session.createObjectMessage();

objectMessage.setStringProperty("key_esteelMessage", object.toString());

return objectMessage;

}

 

public Object fromMessage(Message message) throws JMSException, MessageConversionException {

ObjectMessage objectMessage = (ObjectMessage) message;

return objectMessage.getObjectProperty("key_esteelMessage");

}

 

}