ActiveMQ 和java应用的结合

章节导读

在java应用中嵌入ActiveMQ
ActiveMq结合Spring
用Spring编写JMS客户端

1.1 将ActiveMq嵌入java

首先我们使用JMS API 中的BrokerService类来配置代理,接着我们使用xml文件配置的方式来配置代理.我们通过BrokerFactory类来达到这个目的,

1.1.1 使用org.apache.activemq.broker.BrokerService

这个类用来配置代理以及管理它的生命周期,让我们用之前的认证插件的例子来距离.之前那个例子的xml配置如下:

<broker xmlns="http://activemq.apache.org/schema/core"
brokerName="myBroker"
dataDirectory="${activemq.base}/data">
	<transportConnectors>
		<transportConnector name="openwire"
	uri="tcp://localhost:61616" />
	</transportConnectors>
	<plugins>
		<simpleAuthenticationPlugin>
			<users>
				<authenticationUser username="admin"
	password="password"
	groups="admins,publishers,consumers"/>
				<authenticationUser username="publisher"
	password="password"
	groups="publishers,consumers"/>
				<authenticationUser username="consumer"
	password="password"
	groups="consumers"/>
				<authenticationUser username="guest"
	password="password"
	groups="guests"/>
			</users>
		</simpleAuthenticationPlugin>
	</plugins>
</broker>

对应的java代码如下

import java.util.ArrayList;
import java.util.List;

import org.apache.activemq.broker.BrokerPlugin;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.security.AuthenticationUser;
import org.apache.activemq.security.SimpleAuthenticationPlugin;

public class ActiveMqEmbedJava {
	public static void main(String[] args) throws Exception {
		BrokerService broker = new BrokerService();
		broker.setBrokerName("myBroker");
		broker.setDataDirectory("data/");
		SimpleAuthenticationPlugin authentication = new SimpleAuthenticationPlugin();
		List<AuthenticationUser> users = new ArrayList<AuthenticationUser>();
		users.add(new AuthenticationUser("admin", "password", "admins,publishers,consumers"));
		users.add(new AuthenticationUser("publisher", "password", "publishers,consumers"));
		users.add(new AuthenticationUser("consumer", "password", "consumers"));
		users.add(new AuthenticationUser("guest", "password", "guests"));
		authentication.setUsers(users);
		broker.setPlugins(new BrokerPlugin[] { authentication });
		broker.addConnector("tcp://localhost:61616");
		broker.start();
		System.out.println();
		System.out.println("Press any key to stop the broker");
		System.out.println();
		System.in.read();
	}
}

注意:必须在增加Connector之前配置插件,否则插件不会被初始化.代理启动之后再去增加Connector将不会起到预期的作用.

1.1.2 使用org.apache.activemq.broker.BrokerFactory

在某些情况下,你想用相同的配置初始化代理,就可以使用BrokerFactory.它可以使用ActiveMQ URL来创建一个代理,基于代理 URl协议,找到合适的工厂类并创建对应的BrokerService.用的最多的就是XBeanBrokerFactory,通过XBean风格的URl配置,例如:xbean:/path/to/activemq.xml

这个URl告诉BrokerFactory使用XBeanBrokerFactory以及创建代理实例的路径.BrokerFactory可以使用ActivemQ的xml配置文件来初始化BrokerService.例子如下:

public class Factory {

public static void main(String[] args) throws Exception {
System.setProperty("activemq.base", System.getProperty("user.dir"));
String configUri =
"xbean:target/classes/org/apache/activemq/book/ch6/activemq-simple.xml"
URI brokerUri = new URI(configUri);
BrokerService broker = BrokerFactory.createBroker(brokerUri);
broker.start();
System.out.println();
System.out.println("Press any key to stop the broker");
System.out.println();
System.in.read();
}

1.2 将ActiveMq嵌入Spring

1.纯Spring XMl的形式(将BrokerService 作为一个bean)

spring.xml配置文件,放置在src/main/resources目录下(maven)

<?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:context="http://www.springframework.org/schema/context"
    xmlns:tx="http://www.springframework.org/schema/tx"
	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/tx 
http://www.springframework.org/schema/tx/spring-tx-3.0.xsd
http://www.springframework.org/schema/aop 
http://www.springframework.org/schema/aop/spring-aop-3.0.xsd
">
	<bean id="admins" class="org.apache.activemq.security.AuthenticationUser">
		<constructor-arg index="0" value="admin" />
		<constructor-arg index="1" value="password" />
		<constructor-arg index="2" value="admins,publisher,consumers" />
	</bean>
	<bean id="publishers"
class="org.apache.activemq.security.AuthenticationUser">
		<constructor-arg index="0" value="publisher" />
		<constructor-arg index="1" value="password" />
		<constructor-arg index="2" value="publisher,consumers" />
	</bean>
	<bean id="consumers"
class="org.apache.activemq.security.AuthenticationUser">
		<constructor-arg index="0" value="consumer" />
		<constructor-arg index="1" value="password" />
		<constructor-arg index="2" value="consumers" />
	</bean>
	<bean id="guests" class="org.apache.activemq.security.AuthenticationUser">
		<constructor-arg index="0" value="guest" />
		<constructor-arg index="1" value="password" />
		<constructor-arg index="2" value="guests" />
	</bean>
	<bean id="simpleAuthPlugin"
class="org.apache.activemq.security.SimpleAuthenticationPlugin">
		<property name="users">
		    <list>
		        
				<ref bean="admins" />
				<ref bean="publishers" />
				<ref bean="consumers" />
				<ref bean="guests" />
		    </list>
		</property>
	</bean>
	<bean id="broker" class="org.apache.activemq.broker.BrokerService"
init-method="start" destroy-method="stop">
		<property name="brokerName" value="myBroker" />
		<property name="persistent" value="false" />
		<property name="transportConnectorURIs">
			<list>
				<value>tcp://localhost:61616</value>
			</list>
		</property>
		<property name="plugins">
			<list>
				<ref bean="simpleAuthPlugin"/>
			</list>
		</property>
	</bean>
</beans>

用以下代码加载spring配置文件启动ActiveMq即可

import org.springframework.context.support.ClassPathXmlApplicationContext;

public class PurlSpringXml {

	  public static void main(String[] args) {
		ClassPathXmlApplicationContext cpac = new ClassPathXmlApplicationContext("classpath:/*.xml");
	 }
}

2.使用BrokerFactoryBean

<bean id="broker"
class="org.apache.activemq.xbean.BrokerFactoryBean">
<property name="config"  value="classpath:activemq.xml"/>
<property name="start" value="true" /><!--是否启动一个代理-->
</bean>

3.使用spring的namespace

<beans xmlns="http://www.springframework.org/schema/beans"
	xmlns:amq="http://activemq.apache.org/schema/core" 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-3.0.xsd
http://activemq.apache.org/schema/core
http://activemq.apache.org/schema/core/activemq-core.xsd">
	<amq:broker brokerName="localhost" dataDirectory="${activemq.base}/data">
		<amq:transportConnectors>
			<amq:transportConnector name="openwire"
				uri="tcp://localhost:61616" />
		</amq:transportConnectors>
		<amq:plugins>
			<amq:simpleAuthenticationPlugin>
				<amq:users>
					<amq:authenticationUser username="admin"
						password="password" groups="admins,publishers,consumers" />
					<amq:authenticationUser username="publisher"
						password="password" groups="publishers,consumers" />
					<amq:authenticationUser username="consumer"
						password="password" groups="consumers" />
					<amq:authenticationUser username="guest"
						password="password" groups="guests" />
				</amq:users>
			</amq:simpleAuthenticationPlugin>
		</amq:plugins>
	</amq:broker>
</beans>

1.3 实现请求/应答模式

请求/应答方案包含了应用发送一个消息并按预期的收到那个消息的响应.这样的设计通常都是用server/client模式来实现的,服务器和客户端通过TCP,UDP协议来同步的交流，这种形式的架构伸缩性比较差,并且很难拆分项目.下图就是一个请求应答模式的概述.

客户端和工作者(worker)都是由一个生产者和一个消费者组成的.

首先,生产者创建一个JMS消息格式的请求,并且设置一组重要的属性--关联ID(通过JMSCorrelationID属性),答复的目的地(JMSReplyTo属性).关联id是非常重要的,它允许请求关联到答复上(在有多种请求的情况下).答复的目的地就是回复预期被发送的地方(通常使用临时目的地,资源更友好).客户端配置一个消费者来监听答复目的地.

其次,工作者接受一个请求,处理它,并且发送应答消息,使用在JMSReplyTo属性中设置的目的地,响应消息必须使用原始请求中的JMSCorrelationID来设置应答消息中的JMSCorrelationID属性.当客户端收到应答消息后正确的关联上原始请求.

为了证明它是一种高可可伸缩的方案.假设一个单例的worker不足以承载那么多的请求.只需要增加worker分摊负载即可,这些worker可以分布在不同的主机上.

1.3.1 实现请求响应模式中的代理服务器和worker

/**
 * 请求应答模式,服务器
 * @author cfzhou
 *
 */
public class ActiveMqBrokerAndWorker implements MessageListener {

	private BrokerService broker; 

	private static String brokerURL = "tcp://localhost:61616";//地址

	private Session session;

	private MessageProducer producer;//生产者

	private MessageConsumer consumer;//消费者

	public void start() throws Exception {
		createBroker();
		setupConsumer();
	}
   
	//创建服务
	private void createBroker() throws Exception {
		System.out.println("启动服务");
		broker = new BrokerService();
		broker.setPersistent(false);
		broker.setUseJmx(false);
		broker.addConnector(brokerURL);
		broker.start();
	}

	//启动消费者
	private void setupConsumer() throws JMSException {
		ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(brokerURL);
		Connection connection = connectionFactory.createConnection();
		connection.start();
		session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
		Destination adminQueue = session.createQueue("queue");
		//创建生产者
               //没有默认的目的地，它会发送到每个消息JMSReplyTo属性指定的目的地去
		producer = session.createProducer(null);
		//非持久化
		producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
		//创建消费者
		consumer = session.createConsumer(adminQueue);
		//监听器
		consumer.setMessageListener(this);
	}

	public void stop() throws Exception {
		producer.close();
		consumer.close();
		session.close();
		broker.stop();
	}

	@Override
	public void onMessage(Message message) {
		try {
			//创建回复信息
			TextMessage response = this.session.createTextMessage();
			//处理收到的信息
			if (message instanceof TextMessage) {
				TextMessage txtMsg = (TextMessage) message;
				String messageText = txtMsg.getText();
				response.setText(handleRequest(messageText));
			}
			//设置响应消息对应的关联Id
			System.out.println("收到请求消息Id:" + message.getJMSCorrelationID());
			System.out.println(message.getJMSReplyTo().toString());
                        //设置关联id
			response.setJMSCorrelationID(message.getJMSCorrelationID());
			producer.send(message.getJMSReplyTo(), response);
		} catch (JMSException e) {
			e.printStackTrace();
		}
	}

	public String handleRequest(String messageText) {
		return "Response to '" + messageText + "'";
	}
	
	public static void main(String[] args) throws Exception {
                //开启服务
		ActiveMqBrokerAndWorker  server = new ActiveMqBrokerAndWorker();
		server.start();
		System.out.println();
		System.out.println("Press any key to stop the server");
		System.out.println();
		System.in.read();
		server.stop();
	}
}

1.3.2 实现客户端

/**
 * 请求应答模式,客户端
 * @author cfzhou
 *
 */
public class ActiveMqRequestReplyClient implements MessageListener {

	private static String brokerURL = "tcp://localhost:61616";//地址

	private Session session;

	private MessageProducer producer;//生产者

	private MessageConsumer consumer;//消费者
	
	private  Destination destination;

	private  Destination tempDestination;
	
	private Connection connection;
	
	private AtomicInteger count = new AtomicInteger(0);
	
	public void start() throws Exception {
		setUpClient();	
		
	}
   
    //发送消息
	private void sendMessage() throws JMSException {
		for(int i = 0 ;i< 10 ;i++){
			 Message message = session.createTextMessage("消息" + count.incrementAndGet());  	
			 //设置woker回复的目的地
			 message.setJMSReplyTo(tempDestination);
			 //用UUID生成关联Id
			 String correlationId = UUID.randomUUID().toString();
			 message.setJMSCorrelationID(correlationId);
		     producer.send(destination, message);	
		}
       
	}


	//启动消费者
	private void setUpClient() throws JMSException {
		ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(brokerURL);
		connection = connectionFactory.createConnection();
		connection.start();
		session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
		destination = session.createQueue("queue");
		producer = session.createProducer(destination);
  		producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
  		//创建临时队列
  		tempDestination = session.createTemporaryQueue();
		//创建消费者
		consumer = session.createConsumer(tempDestination);
		//监听器
		consumer.setMessageListener(this);
	}

	public void stop() throws Exception {
		producer.close();
		consumer.close();
		session.close();		
		connection.close();
	}

	@Override
	public void onMessage(Message message) {
		try {			
			//处理收到的信息			
			TextMessage txtMsg = (TextMessage) message;
			String messageText = txtMsg.getText();
			System.out.println("收到答复消息:" + messageText + ",messageid:" + message.getJMSMessageID());
		} catch (JMSException e) {
			e.printStackTrace();
		}
	}

	public String handleRequest(String messageText) {
		return " to '" + messageText + "'";
	}
	
	public static void main(String[] args) throws Exception {
		ActiveMqRequestReplyClient client = new ActiveMqRequestReplyClient();
		client.start();
		client.sendMessage();
		//Thread.sleep(3000);
		//client.stop();
	}
}

客户端的消费者发送消息到请求队列,消费者监听临时队列.

1.4 用spring写JMS客户端

1.配置JMS连接

<bean id="jmsConnectionFactory"
class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://localhost:61616" />
<property name="userName" value="admin" />
<property name="password" value="password" />
</bean>

通常为了性能还是需要配置连接池,activeMq提供了PooledConnectionFactory类,它提供了JMS连接池和session.

2.配置连接池

<bean id="pooledJmsConnectionFactory"
class="org.apache.activemq.pool.PooledConnectionFactory"
destroy-method="stop">
<property name="connectionFactory" ref="jmsConnectionFactory" />
</bean>

连接池maven

<!-- https://mvnrepository.com/artifact/org.apache.activemq/activemq-pool -->
<dependency>
    <groupId>org.apache.activemq</groupId>
    <artifactId>activemq-pool</artifactId>
    <version>5.14.3</version>
</dependency>

3.配置Destinations(队列和主题)

<bean id="cscoDest" class="org.apache.activemq.command.ActiveMQQueue">
    <constructor-arg value="STOCKS.CSCO" />
</bean>
<bean id="orclDest" class="org.apache.activemq.command.ActiveMQTopic">
    <constructor-arg value="STOCKS.ORCL" />
</bean>

4. 配置消费者

<!-- 消息监听器-->
<bean id="portfolioListener" class="com.zcf.activemq.chapter7.spring.JmsMessageListener">
</bean>
<!-- Spring  队列消费者 -->
<bean id="cscoConsumer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
  <property name="connectionFactory" ref="jmsConnectionFactory" />
  <property name="destination" ref="cscoDest" />
   <property name="messageListener" ref="portfolioListener" />
</bean>
<!-- Spring 主题消费者 -->
<bean id="orclConsumer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
  <property name="connectionFactory" ref="jmsConnectionFactory" />
  <property name="destination" ref="orclDest" />
  <property name="messageListener" ref="portfolioListener" />
</bean>

消息监听器

public class JmsMessageListener implements MessageListener{

	@Override
	public void onMessage(Message message) {
		MapMessage map = (MapMessage) message;
	}

}

5.配置生产者

spring也提供了发送消息的模板,JmsTemplate. 使用spring发送消息最普遍的方式就是靠实现Spring MessageCreator接口.这个接口只定义了一个方法createMessage(),返回一个jms消息,

下面的例子中,实现了创建一个随机价格逻辑的消息,并且创建了一个MapMessage来保存多个数据.为了发送消息,需要利用JmsTemplate’s send()方法.

public class StockMessageCreator implements MessageCreator {
	private int MAX_DELTA_PERCENT = 1;
	private Map<Destination, Double> LAST_PRICES = new Hashtable<Destination, Double>();
	Destination stock;

	public StockMessageCreator(Destination stock) {
		this.stock = stock;
	}

	public Message createMessage(Session session) throws JMSException {
		Double value = LAST_PRICES.get(stock);
		if (value == null) {
			value = new Double(Math.random() * 100);
		}
		// lets mutate the value by some percentage
		double oldPrice = value.doubleValue();
		value = new Double(mutatePrice(oldPrice));
		LAST_PRICES.put(stock, value);
		double price = value.doubleValue();
		double offer = price * 1.001;
		boolean up = (price > oldPrice);
		MapMessage message = session.createMapMessage();
		message.setString("stock", stock.toString());
		message.setDouble("price", price);
		message.setDouble("offer", offer);
		message.setBoolean("up", up);
		System.out.println("Sending: " + ((ActiveMQMapMessage) message).getContentMap() + " on destination: " + stock);
		return (Message) message;
	}
	protected double mutatePrice(double price) {
		double percentChange = (2 * Math.random() * MAX_DELTA_PERCENT)
		- MAX_DELTA_PERCENT;
		return price * (100 + percentChange) / 100;
	}
	
}

使用spring Jms模板发送消息的类

public class SpringPublisher {
	//JMS模板
	private JmsTemplate template;
	private int count = 10;
	private int total;
	private Destination[] destinations;
	private HashMap<Destination, StockMessageCreator> creators = new HashMap<Destination, StockMessageCreator>();

	public void start() {
		while (total < 1000) {
			for (int i = 0; i < count; i++) {
				sendMessage();
			}
			total += count;
			System.out.println("Published '" + count + "' of '" + total + "' price messages");
			try {
				Thread.sleep(1000);
			} catch (InterruptedException x) {
			}
		}
	}

	protected void sendMessage() {
		int idx = 0;
		while (true) {
			idx = (int) Math.round(destinations.length * Math.random());
			if (idx < destinations.length) {
				break;
			}
		}
		Destination destination = destinations[idx];
		template.send(destination, getStockMessageCreator(destination));
	}

	private StockMessageCreator getStockMessageCreator(Destination dest) {
		if (creators.containsKey(dest)) {
			return creators.get(dest);
		} else {
			StockMessageCreator creator = new StockMessageCreator(dest);
			creators.put(dest, creator);
			return creator;
		}
	}
	// getters
}

配置jms模板和发送消息类到spring

<!-- Spring JMS Template -->
	<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
		<property name="connectionFactory" ref="pooledJmsConnectionFactory" />
	</bean>
	<bean id="stockPublisher" class="com.zcf.activemq.chapter7.spring.SpringPublisher">
		<property name="template" ref="jmsTemplate" />
		<property name="destinations">
			<list>
				<ref local="cscoDest" />
				<ref local="orclDest" />
			</list>
		</property>
	</bean>

测试类

public class SpringClient {
	public static void main(String[] args) throws Exception {
		BrokerService broker = new BrokerService();
		broker.addConnector("tcp://localhost:61616");
		broker.setPersistent(false);
		broker.start();
		  ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("classpath:/spring-jmstemplate.xml");
		SpringPublisher publisher = (SpringPublisher) context.getBean("stockPublisher");
		publisher.start();
	}
}

第八章 - ActiveMQ 和java应用的结合

ActiveMQ 和java应用的结合

章节导读

相关推荐