activeMq配置

基本发送与接收

applicationContext.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"
xsi:schemaLocation="http://www.springframework.org/schema/beans


http://www.springframework.org/schema/beans/spring-beans-2.0.xsd">
<!--创建连接工厂 -->
<bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://localhost:61616"></property>
</bean>
<!--创建主题连接工厂 -->
<bean id="topicListenConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://127.0.0.1:61616" />
</bean>
<!--创建主题 -->
<bean id="myTopic" class="org.apache.activemq.command.ActiveMQTopic">
<constructor-arg value="normandy.topic"/>
</bean>
<!--消息转换器 -->
<bean id="messageConvertForSys" class="com.normandy.tech.test.MessageConvertForSys">

<bean id="topicSendJmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory" ref="topicSendConnectionFactory"></property>
<property name="defaultDestination" ref="myTopic" />
<property name="messageConverter" ref="messageConvertForSys" />
<!-- 开启订阅模式 -->
<property name="pubSubDomain" value="true"/>
</bean>
<!-- 声明ActiveMQ消息目标,目标可以是一个队列，也可以是一个主题ActiveMQTopic -->
<bean id="destination" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg index="0" value="rantz.marketing.queue"></constructor-arg>
</bean>
<!---->
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory" ref="connectionFactory"></property>
<property name="defaultDestination" ref="destination"></property>
<property name="receiveTimeout" value="6000"></property>


</bean>
<bean id="sender" class="com.roadrantz.marketing.Sender">
<property name="jmsTemplate" ref="jmsTemplate"></property>


</bean>
<bean id="receiver" class="com.roadrantz.marketing.Receiver">
<property name="jmsTemplate" ref="jmsTemplate"></property>
</bean>
</beans>

生产者

public class Sender {
private JmsTemplate jmsTemplate;


public JmsTemplate getJmsTemplate() {
return jmsTemplate;
}


public void setJmsTemplate(JmsTemplate jmsTemplate) {
this.jmsTemplate = jmsTemplate;
}


public void sendInfo() {
this.jmsTemplate.send(new MessageCreator() {


@Override
public Message createMessage(Session arg0) throws JMSException {
MapMessage mess = arg0.createMapMessage();
mess.setString("lastname", "pp");
return mess;
}


});
}
}

消费者

public class Receiver {
private JmsTemplate jmsTemplate;


public JmsTemplate getJmsTemplate() {
return jmsTemplate;
}


public void setJmsTemplate(JmsTemplate jmsTemplate) {
this.jmsTemplate = jmsTemplate;
}
public Receiver(){}
public String receiveMessage() throws JMSException{
MapMessage mess=(MapMessage) jmsTemplate.receive();
String re=mess.getString("lastname");
return re;
}
}

持久化

将mysql的jar包拷贝到active的lib目录下

修改activemq/conf/activemq.xml文件

增加数据源配置

<bean id="mysql-ds" class="org.apache.commons.dbcp.BasicDataSource" destroy-method="close">
<property name="driverClassName" value="com.mysql.jdbc.Driver"/>
<property name="url" value="jdbc:mysql://localhost/mysql?relaxAutoCommit=true"/>
<property name="username" value="root"/>
<property name="password" value="root"/>
<property name="poolPreparedStatements" value="true"/>
</bean>

持久化适配器

<persistenceAdapter>
<!-- <journaledJDBC journalLogFiles="5" dataDirectory="${activemq.base}/activemq-data"/> -->
<!-- To use a different datasource, use the following syntax : -->

<journaledJDBC journalLogFiles="5" dataDirectory="../activemq-data" dataSource="#mysql-ds"/>

</persistenceAdapter>

定义转换器

public class MessageConvertForSys implements MessageConverter {
/* (non-Javadoc)
* @see org.springframework.jms.support.converter.MessageConverter#toMessage(java.lang.Object, javax.jms.Session)
*/
public Message toMessage(Object object, Session session)
throws JMSException, MessageConversionException {
System.out.println("[toMessage]");
ObjectMessage objectMessage = session.createObjectMessage();

objectMessage.setJMSExpiration(1000);
objectMessage.setStringProperty("key1", object+"_add");

return objectMessage;
}
/* (non-Javadoc)
* @see org.springframework.jms.support.converter.MessageConverter#fromMessage(javax.jms.Message)
*/
public Object fromMessage(Message message) throws JMSException,
MessageConversionException {
System.out.println("[fromMessage]");
ObjectMessage objectMessage = (ObjectMessage) message;

return objectMessage.getObjectProperty("key1");
}
}

spring消息驱动pojo spring+activeMq+jencks

jencks是一个JCA容器，可以控制mq连接，据说可以控制事务

applicationContext

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans


http://www.springframework.org/schema/beans/spring-beans-2.0.xsd">
<bean id="helloBean" class="mdp.demo.HelloBean" />
<bean id="jmsTransactionManager" class="org.jencks.factory.TransactionManagerFactoryBean" />
<bean id="activeMQContainer" class="org.jencks.JCAContainer">
<property name="bootstrapContext">
<bean id="bootstrapContext" class="org.jencks.factory.BootstrapContextFactoryBean">
<property name="transactionManager" ref="jmsTransactionManager" />
</bean>
</property>


<property name="resourceAdapter">
<bean id="activeMQResourceAdapter" class="org.apache.activemq.ra.ActiveMQResourceAdapter">
<property name="serverUrl" value="tcp://localhost:61616" />
</bean>
</property>
</bean>
<bean id="HelloMDP" class="org.jencks.JCAConnector">
<property name="jcaContainer" ref="activeMQContainer" />
<property name="transactionManager" ref="jmsTransactionManager" />
<property name="activationSpec">
<bean class="org.apache.activemq.ra.ActiveMQActivationSpec">
<property name="destination" value="Hello.Queue" />
<property name="destinationType" value="javax.jms.Queue" />
</bean>
</property>
<property name="ref" value="helloBean" />
</bean>
</beans>

监听类

public class HelloBean implements MessageListener {


@Override
public void onMessage(Message arg0) {
// TODO Auto-generated method stub
try {
String name=arg0.getStringProperty("name");
System.out.println(name);
} catch (JMSException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}

}


}

客户端applicationContext.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"
xsi:schemaLocation="http://www.springframework.org/schema/beans


http://www.springframework.org/schema/beans/spring-beans-2.0.xsd">
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<property name="defaultDestinationName" value="messages.input" />
<property name="connectionFactory" ref="connectionFactory" />
</bean>
<bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://localhost:61616" />
</bean>
</beans>

服务端启动

public class ServerTest {


/**
* @param args
*/
public static void main(String[] args) {
// TODO Auto-generated method stub
ApplicationContext ctx=new ClassPathXmlApplicationContext("applicationContext.xml");

}


}

客户端

package mdp.demo;


import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;


import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;


import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;


public class MDPClient {


/**
* @param args
*/
public static void main(String[] args) throws Exception{
// TODO Auto-generated method stub
ApplicationContext ctx=new ClassPathXmlApplicationContext("spring-client.xml");
JmsTemplate tmp=(JmsTemplate) ctx.getBean("jmsTemplate");
tmp.setDefaultDestinationName("Hello.Queue");
tmp.send(new MessageCreator(){


@Override
public Message createMessage(Session arg0) throws JMSException {
// TODO Auto-generated method stub
Message mes=arg0.createMapMessage();
mes.setStringProperty("name", "hello");
return mes;
}

});
}


}

事务

<beans>
<!-- ActiveMQ Broker -->
<bean id="broker" class="org.apache.activemq.broker.BrokerService" init-method="start" destroy-method="stop">
<property name="persistent" value="false"/>
<property name="transportConnectorURIs">
<list>
<value>tcp://localhost:5000</value>
</list>
</property>
</bean>


<!-- Geronimo Transaction Manager -->
<bean id="transactionContextManager" class="org.jencks.factory.TransactionContextManagerFactoryBean"/>
<bean id="geronimo" class="org.jencks.factory.GeronimoTransactionManagerFactoryBean"/>
<bean id="geronimoTransactionManager" class="org.springframework.transaction.jta.JtaTransactionManager">
<property name="userTransaction" ref="geronimo" />
</bean>


<!-- Jencks Connection Manager -->
<bean id="connectionManager" class="org.jencks.factory.ConnectionManagerFactoryBean">
<property name="transactionSupport">
<bean class="org.jencks.factory.XATransactionFactoryBean">
<property name="useTransactionCaching" value="true"/>
<property name="useThreadCaching" value="false"/>
</bean>
</property>
<property name="poolingSupport">
<bean class="org.jencks.factory.SinglePoolFactoryBean">
<property name="maxSize" value="2"/>
<property name="minSize" value="1"/>
<property name="blockingTimeoutMilliseconds" value="60"/>
<property name="idleTimeoutMinutes" value="60"/>
<property name="matchOne" value="true"/>
<property name="matchAll" value="true"/>
<property name="selectOneAssumeMatch" value="true"/>
</bean>
</property>
</bean>


<!-- ActiveMQ Connection -->
<bean id="jmsResourceAdapter" class="org.apache.activemq.ra.ActiveMQResourceAdapter" depends-on="broker">
<property name="serverUrl">
<value>tcp://localhost:5000</value>
</property>
</bean>
<bean id="jmsManagedConnectionFactory" class="org.apache.activemq.ra.ActiveMQManagedConnectionFactory">
<property name="resourceAdapter" ref="jmsResourceAdapter"/>
</bean>
<bean id="jmsConnectionFactory" class="org.springframework.jca.support.LocalConnectionFactoryBean">
<property name="managedConnectionFactory" ref="jmsManagedConnectionFactory"/>
<property name="connectionManager" ref="connectionManager"/>
</bean>


<!-- Tranql JDBC Connection -->
<!--
<bean id="tranqlManagedConnectionFactory" class="org.jencks.tranql.XAPoolDataSourceMCF">
<property name="driverName" value="org.postgresql.Driver"/>
<property name="url" value="jdbc:postgresql://ats-manager/activemq"/>
<property name="user" value="activemq"/>
</bean>
<bean id="tranqlDataSource" class="org.springframework.jca.support.LocalConnectionFactoryBean">
<property name="managedConnectionFactory" ref="tranqlManagedConnectionFactory"/>
<property name="connectionManager" ref="connectionManager"/>
</bean>
-->


<!-- Enhydra JDBC Connection -->
<bean id="enhydraDataSource" class="org.enhydra.jdbc.pool.StandardXAPoolDataSource" destroy-method="shutdown">
<property name="dataSource">
<bean class="org.enhydra.jdbc.standard.StandardXADataSource" destroy-method="shutdown">
<property name="transactionManager" ref="geronimo" />
<property name="driverName" value="org.postgresql.Driver" />
<property name="url" value="jdbc:postgresql://ats-manager/activemq" />
</bean>
</property>
<property name="user" value="activemq"/>
</bean>


<bean id="jencksJCAContainer" class="org.jencks.JCAContainer">
<property name="bootstrapContext">
<bean class="org.jencks.factory.BootstrapContextFactoryBean">
<property name="threadPoolSize" value="25"/>
</bean>
</property>
<property name="resourceAdapter" ref="jmsResourceAdapter"/>
</bean>


<bean id="inboundConnector" class="org.jencks.JCAConnector">
<property name="jcaContainer" ref="jencksJCAContainer" />
<property name="activationSpec">
<bean class="org.apache.activemq.ra.ActiveMQActivationSpec">
<property name="destination" value="messages.input"/>
<property name="destinationType" value="javax.jms.Queue"/>
</bean>
</property>
<property name="transactionManager" ref="geronimo"/>
<property name="ref" value="echoBean"/>
</bean>


<bean id="echoBean" class="transactions.EchoBean">
<property name="jdbcTemplate">
<bean class="org.springframework.jdbc.core.JdbcTemplate">
<property name="dataSource" ref="enhydraDataSource"/>
</bean>
</property>
<property name="jmsTemplate">
<bean class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory" ref="jmsConnectionFactory"/>
</bean>
</property>
</bean>
</beans>

public class EchoBean implements MessageListener {


private Log log = LogFactory.getLog(getClass());


private JdbcTemplate jdbcTemplate;
private JmsTemplate jmsTemplate;


public void setJdbcTemplate(JdbcTemplate jdbcTemplate) {
this.jdbcTemplate = jdbcTemplate;
}


public void setJmsTemplate(JmsTemplate jmsTemplate) {
this.jmsTemplate = jmsTemplate;
}


public void onMessage(Message message) {
log.debug(message);
if (message instanceof TextMessage) {
try {
String messageText = ((TextMessage)message).getText();
log.debug("execute JMS operation");
jmsTemplate.convertAndSend("messages.ouptut", messageText);
log.debug("execute JDBC operation");
jdbcTemplate.execute("insert into t1 values('"+messageText+"')");
} catch (JMSException e) {
e.printStackTrace();
}
}
}
}