JMS Active MQ의 스프링 통합에 대한 간략한 분석
- 巴扎黑원래의
- 2017-06-23 10:10:521454검색
1. ptp의 동기식 메시지 수신을 실현하기 위해 spring과 통합
pom.xml:
<!-- --><dependency><groupId>org.springframework</groupId><artifactId>spring-jms</artifactId><version>4.3.7.RELEASE</version></dependency> <!-- --><dependency><groupId>org.apache.activemq</groupId><artifactId>activemq-pool</artifactId><version>5.9.0</version></dependency>
spring-jms.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:jms="http://www.springframework.org/schema/jms"xmlns:amq="http://activemq.apache.org/schema/core"xsi:schemaLocation="http://activemq.apache.org/schema/core http://www.springframework.org/schema/jms http://www.springframework.org/schema/beans "><!-- ActiveMQConnectionFactory就是JMS中负责创建到ActiveMQ连接的工厂类 --><bean id="connectionFactory" class="org.apache.activemq.spring.ActiveMQConnectionFactory" > <property name="brokerURL" value="tcp://192.168.0.224:61616"/> </bean><!-- 创建连接池 --><bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop"> <property name="connectionFactory" ref="connectionFactory"/> <property name="maxConnections" value="10"/> </bean> <!-- Spring为我们提供了多个ConnectionFactory，有SingleConnectionFactory和CachingConnectionFactory --><bean id="cachingConnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory"> <property name="targetConnectionFactory" ref="pooledConnectionFactory"/> </bean> <!-- Spring提供的JMS工具类，它可以进行消息发送、接收等 --> <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate"> <!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 --> <property name="connectionFactory" ref="cachingConnectionFactory"/> </bean> <!--这个是队列目的地，点对点的--> <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue"> <constructor-arg index="0" value="spring-queue"/> </bean> </beans>
ConnectionFactory는 JMS 서버에 대한 링크를 생성하는 데 사용됩니다. Spring은 우리는 SingleConnectionFactory 및 CachingConnectionFactory를 포함한 여러 ConnectionFactory를 제공합니다. SingleConnectionFactory는 JMS 서버 링크 설정 요청에 대해 항상 동일한 링크를 반환하고 Connection의 닫기 메소드 호출을 무시합니다. CachingConnectionFactory는 SingleConnectionFactory를 상속받아 SingleConnectionFactory의 모든 기능을 가지며, Session, MessageProducer, MessageConsumer를 캐시할 수 있는 새로운 캐싱 기능도 추가합니다. 여기서는 CachingConnectionFactory를 예로 사용합니다.
Message producer:
= ClassPathXmlApplicationContext("spring-jms.xml"=(JmsTemplate) context.getBean("jmsTemplate"=(Destination) context.getBean("queueDestination" Message createMessage(Session session) session.createTextMessage("Hello spring JMS"Consumer:
package com.jalja.org.jms.spring;import javax.jms.Destination;import org.springframework.context.ApplicationContext;import org.springframework.context.support.ClassPathXmlApplicationContext;import org.springframework.jms.core.JmsTemplate;public class SpringJmsReceive {public static void main(String[] args) {
ApplicationContext context=new ClassPathXmlApplicationContext("spring-jms.xml");
JmsTemplate jmsTemplate=(JmsTemplate) context.getBean("jmsTemplate");
Destination queueDestination=(Destination) context.getBean("queueDestination");
String msg=(String) jmsTemplate.receiveAndConvert(queueDestination);
System.out.println(msg);
}
}2. PTP의 비동기 호출
Spring에서 메시지를 비동기적으로 수신하도록 리스너를 직접 구성합니다. 이는 Spring에서 구성하는 것과 동일합니다. 소비자인 경우 메시지를 받을 때 소비자를 시작할 필요가 없습니다.
spring-jms.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:jms="http://www.springframework.org/schema/jms"xmlns:amq="http://activemq.apache.org/schema/core"xsi:schemaLocation="http://activemq.apache.org/schema/core http://www.springframework.org/schema/jms http://www.springframework.org/schema/beans "><!-- ActiveMQConnectionFactory就是JMS中负责创建到ActiveMQ连接的工厂类 --><bean id="connectionFactory" class="org.apache.activemq.spring.ActiveMQConnectionFactory" > <property name="brokerURL" value="tcp://192.168.0.224:61616"/> </bean><!-- 创建连接池 --><bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop"> <property name="connectionFactory" ref="connectionFactory"/> <property name="maxConnections" value="10"/> </bean> <!-- Spring为我们提供了多个ConnectionFactory，有SingleConnectionFactory和CachingConnectionFactory --><bean id="cachingConnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory"> <property name="targetConnectionFactory" ref="pooledConnectionFactory"/> </bean> <!-- Spring提供的JMS工具类，它可以进行消息发送、接收等 --> <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate"> <!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 --> <property name="connectionFactory" ref="cachingConnectionFactory"/> </bean> <!--这个是队列目的地，点对点的--> <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue"> <constructor-arg index="0" value="spring-queue"/> </bean> <!-- 消息监听器 --> <bean id="myMessageListener" class="com.jalja.org.jms.spring.yb.MyMessageListener"/> <!-- 消息监听容器 --> <bean id="jmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer"> <property name="connectionFactory" ref="cachingConnectionFactory" /> <property name="destination" ref="queueDestination" /> <property name="messageListener" ref="myMessageListener" /> </bean> </beans>
생산자가 지정된 대상 대상으로 메시지를 보낸 후 다음 단계는 소비자가 지정된 대상에서 메시지를 소비하는 것입니다. 그렇다면 소비자는 생산자가 지정된 대상으로 메시지를 보냈다는 것을 어떻게 알 수 있습니까? 이는 Spring이 캡슐화하는 메시지 수신 컨테이너 MessageListenerContainer를 통해 달성됩니다. 이는 정보를 수신하고 수신된 정보를 처리를 위해 실제 MessageListener에 배포하는 역할을 합니다. 각 소비자는 각 대상에 해당하는 MessageListenerContainer가 필요합니다. 메시지 수신 컨테이너의 경우 수신할 대상을 아는 것 외에도 수신할 위치도 알아야 합니다. 즉, 수신할 JMS 서버도 알아야 합니다. 이는 MessageConnectionFactory를 삽입하여 수행됩니다. 이를 달성하기 위해 MessageConnectionFactory를 구성할 때. 따라서 MessageListenerContainer를 구성할 때 지정해야 하는 세 가지 속성이 있습니다. 하나는 수신할 위치를 나타내는 ConnectionFactory이고, 다른 하나는 수신할 메시지를 처리하는 MessageListener입니다. 메시지. Spring은 SimpleMessageListenerContainer와 DefaultMessageListenerContainer라는 두 가지 유형의 MessageListenerContainer를 제공합니다.
SimpleMessageListenerContainer: SimpleMessageListenerContainer는 처음에 세션과 소비자를 생성하고 표준 JMS MessageConsumer.setMessageListener() 메서드를 사용하여 리스너를 등록하고 JMS 공급자가 리스너의 콜백 함수를 호출하도록 합니다. 런타임 요구 사항에 동적으로 적응하지 않으며 외부 트랜잭션 관리에 참여하지 않습니다. 호환성 측면에서는 독립형 JMS 사양과 매우 유사하지만 일반적으로 Java EE의 JMS 제한 사항과 호환되지 않습니다.
DefaultMessageListenerContainer: 대부분의 경우 여전히 DefaultMessageListenerContainer를 사용합니다. SimpleMessageListenerContainer와 비교하여 DefaultMessageListenerContainer는 런타임 요구 사항에 동적으로 적응하고 외부 트랜잭션 관리에 참여할 수 있습니다. 이는 JMS 제공자에 대한 낮은 요구 사항, 트랜잭션 참여와 같은 고급 기능 및 Java EE 환경과의 호환성 사이에서 균형을 잘 이루고 있습니다.
메시지 생성자:
public static void main(String[] args) {
ApplicationContext context=new ClassPathXmlApplicationContext("spring-jms.xml");
JmsTemplate jmsTemplate=(JmsTemplate) context.getBean("jmsTemplate");
Destination queueDestination=(Destination) context.getBean("queueDestination");
System.out.println("异步调用执行开始");
jmsTemplate.send(queueDestination, new MessageCreator(){
@Overridepublic Message createMessage(Session session) throws JMSException {return session.createTextMessage("Hello spring JMS");
}
});
System.out.println("异步调用执行结束");
}메시지 리스너: MyMessageListener
public class MyMessageListener implements MessageListener{
@Overridepublic void onMessage(Message message) {
TextMessage msg= (TextMessage) message;try {
System.out.println("你好："+msg.getText());
} catch (JMSException e) {
e.printStackTrace();
}
}
}메시지 생성자를 시작합니다. 리스너의 실행 결과는
异步调用执行开始 异步调用执行结束 你好：Hello spring JMS
3입니다. -jms .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:jms="http://www.springframework.org/schema/jms"xmlns:amq="http://activemq.apache.org/schema/core"xsi:schemaLocation="http://activemq.apache.org/schema/core http://www.springframework.org/schema/jms http://www.springframework.org/schema/beans "><!-- ActiveMQConnectionFactory就是JMS中负责创建到ActiveMQ连接的工厂类 --><bean id="connectionFactory" class="org.apache.activemq.spring.ActiveMQConnectionFactory" > <property name="brokerURL" value="tcp://192.168.0.224:61616"/> </bean><!-- 创建连接池 --><bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop"> <property name="connectionFactory" ref="connectionFactory"/> <property name="maxConnections" value="10"/> </bean> <!-- Spring为我们提供了多个ConnectionFactory，有SingleConnectionFactory和CachingConnectionFactory --><bean id="cachingConnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory"> <property name="targetConnectionFactory" ref="pooledConnectionFactory"/> </bean> <!-- Spring提供的JMS工具类，它可以进行消息发送、接收等 --> <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate"> <!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 --> <property name="connectionFactory" ref="cachingConnectionFactory"/> </bean> <!--这个是队列目的地，发布订阅--> <bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic"> <constructor-arg index="0" value="spring-Topic"/> </bean> </beans>생산자:
public static void main(String[] args) {
ApplicationContext context=new ClassPathXmlApplicationContext("spring-jms.xml");
JmsTemplate jmsTemplate=(JmsTemplate) context.getBean("jmsTemplate");
Destination topicDestination=(Destination) context.getBean("topicDestination");
jmsTemplate.send(topicDestination, new MessageCreator(){
@Overridepublic Message createMessage(Session session) throws JMSException {return session.createTextMessage("Hello spring JMS topicDestination");
}
});
}
소비자:
public class SpringJmsSubscriber {public static void main(String[] args) {
ApplicationContext context=new ClassPathXmlApplicationContext("spring-jms.xml");
JmsTemplate jmsTemplate=(JmsTemplate) context.getBean("jmsTemplate");
Destination topicDestination=(Destination) context.getBean("topicDestination");
String msg=(String) jmsTemplate.receiveAndConvert(topicDestination);
System.out.println(msg);
}
}
위 내용은 JMS Active MQ의 스프링 통합에 대한 간략한 분석의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!