怎么开始Spring 集成

本文介绍如何在 Spring 框架下用 MQ 收发消息。主要包括三部分内容：普通消息生产者和 Spring 集成，事务消息生产者和 Spring 集成， 消息消费者与 Spring 集成。
请确保同一个 Consumer ID 下所有 Consumer 实例的订阅关系保持一致，具体请参考 订阅关系一致文档。
Spring 框架下支持的配置参数和 TCP Java 一致，具体可以参考 Java SDK 使用说明。

生产者与 Spring 集成

1. 
   在 producer.xml 中定义生产者 Bean 等信息。 <?xml version="1.0" encoding="UTF-8"?>
2. <beans xmlns="http://www.springframework.org/schema/beans"
3.         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
4.         xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
6.      <bean id="producer" class="com.aliyun.openservices.ons.api.bean.ProducerBean" init-method="start" destroy-method="shutdown">
7.      <!-- Spring接入方式支持Java SDK支持的所有配置项 -->
8.          <property name="properties" > <!--生产者配置信息-->
9.              <props>
10.                  <prop key="ProducerId">PID_DEMO</prop> <!--请替换XXX-->
11.                  <prop key="AccessKey">XXX</prop>
12.                  <prop key="SecretKey">XXX</prop>
13.              </props>
14.          </property>
15.      </bean>
17. </beans>

1. package demo;
3. import com.aliyun.openservices.ons.api.Message;
4. import com.aliyun.openservices.ons.api.Producer;
5. import com.aliyun.openservices.ons.api.SendResult;
6. import com.aliyun.openservices.ons.api.exception.ONSClientException;
7. import org.springframework.context.ApplicationContext;
8. import org.springframework.context.support.ClassPathXmlApplicationContext;
10. public class ProduceWithSpring {
11.      public static void main(String[] args) {
12.          /**
13.           * 生产者Bean配置在producer.xml中,可通过ApplicationContext获取或者直接注入到其他类(比如具体的Controller)中.
14.           */
15.          ApplicationContext context = new ClassPathXmlApplicationContext("producer.xml");
17.          Producer producer = (Producer) context.getBean("producer");
18.          //循环发送消息
19.          for (int i = 0; i < 100; i++) {
20.              Message msg = new Message( //
21.                      // Message所属的Topic
22.                      "TopicTestMQ",
23.                      // Message Tag 可理解为Gmail中的标签，对消息进行再归类，方便Consumer指定过滤条件在MQ服务器过滤
24.                      "TagA",
25.                      // Message Body 可以是任何二进制形式的数据， MQ不做任何干预
26.                      // 需要Producer与Consumer协商好一致的序列化和反序列化方式
27.                      "Hello MQ".getBytes());
28.              // 设置代表消息的业务关键属性，请尽可能全局唯一
29.              // 以方便您在无法正常收到消息情况下，可通过MQ 控制台查询消息并补发
30.              // 注意：不设置也不会影响消息正常收发
31.              msg.setKey("ORDERID_100");
32.              // 发送消息，只要不抛异常就是成功
33.              try {
34.                  SendResult sendResult = producer.send(msg);
35.                  assert sendResult != null;
36.                  System.out.println("send success: " + sendResult.getMessageId());
37.              }catch (ONSClientException e) {
38.                  System.out.println("发送失败");
39.              }
41.          }
42.      }
43. }

1. 
   首先需要实现一个 LocalTransactionChecker，如下所示。一个消息生产者只能有一个 LocalTransactionChecker。 package demo;
3. import com.aliyun.openservices.ons.api.Message;
4. import com.aliyun.openservices.ons.api.transaction.LocalTransactionChecker;
5. import com.aliyun.openservices.ons.api.transaction.TransactionStatus;
7. public class DemoLocalTransactionChecker implements LocalTransactionChecker {
8.      public TransactionStatus check(Message msg) {
9.          System.out.println("开始回查本地事务状态");
10.          return TransactionStatus.CommitTransaction; //根据本地事务状态检查结果返回不同的TransactionStatus
11.      }
12. }

1. <?xml version="1.0" encoding="UTF-8"?>
2. <beans xmlns="http://www.springframework.org/schema/beans"
3.         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
4.         xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
6.      <bean id="localTransactionChecker" class="demo.DemoLocalTransactionChecker"></bean>
8.      <bean id="transactionProducer" class="com.aliyun.openservices.ons.api.bean.TransactionProducerBean" init-method="start" destroy-method="shutdown">
9.          <property name="properties" > <!--事务消息生产者配置信息-->
10.              <props>
11.                  <prop key="ProducerId">PID_DEMO</prop> <!--请替换XXX-->
12.                  <prop key="AccessKey">AKDEMO</prop>
13.                  <prop key="SecretKey">SKDEMO</prop>
14.              </props>
15.          </property>
16.          <property name="localTransactionChecker" ref="localTransactionChecker"></property>
17.      </bean>
19. </beans>

1. package demo;
3. import com.aliyun.openservices.ons.api.Message;
4. import com.aliyun.openservices.ons.api.SendResult;
5. import com.aliyun.openservices.ons.api.transaction.LocalTransactionExecuter;
6. import com.aliyun.openservices.ons.api.transaction.TransactionProducer;
7. import com.aliyun.openservices.ons.api.transaction.TransactionStatus;
8. import org.springframework.context.ApplicationContext;
9. import org.springframework.context.support.ClassPathXmlApplicationContext;
11. public class ProduceTransMsgWithSpring {
13.      public static void main(String[] args) {
14.          /**
15.           * 事务消息生产者Bean配置在transactionProducer.xml中,可通过ApplicationContext获取或者直接注入到其他类(比如具体的Controller)中.
16.           * 请结合例子"发送事务消息"
17.           */
18.          ApplicationContext context = new ClassPathXmlApplicationContext("transactionProducer.xml");
20.          TransactionProducer transactionProducer = (TransactionProducer) context.getBean("transactionProducer");
22.          Message msg = new Message("XXX", "TagA", "Hello MQ transaction===".getBytes());
24.          SendResult sendResult = transactionProducer.send(msg, new LocalTransactionExecuter() {
25.              @Override
26.              public TransactionStatus execute(Message msg, Object arg) {
27.                  System.out.println("执行本地事务");
28.                  return TransactionStatus.CommitTransaction; //根据本地事务执行结果来返回不同的TransactionStatus
29.              }
30.          }, null);
31.      }
32. }

1. 
   创建 MessageListener，如下所示。 package demo;
3. import com.aliyun.openservices.ons.api.Action;
4. import com.aliyun.openservices.ons.api.ConsumeContext;
5. import com.aliyun.openservices.ons.api.Message;
6. import com.aliyun.openservices.ons.api.MessageListener;
8. public class DemoMessageListener implements MessageListener {
10.      public Action consume(Message message, ConsumeContext context) {
12.          System.out.println("Receive: " + message.getMsgID());
13.          try {
14.              //do something..
15.              return Action.CommitMessage;
16.          }catch (Exception e) {
17.              //消费失败
18.              return Action.ReconsumeLater;
19.          }
20.      }
21. }

1. <?xml version="1.0" encoding="UTF-8"?>
2. <beans xmlns="http://www.springframework.org/schema/beans"
3.         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
4.         xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
6.      <bean id="msgListener" class="demo.DemoMessageListener"></bean> <!--Listener配置-->
7. <!-- 多CID订阅同一个Topic，可以创建多个ConsumerBean-->
8.      <bean id="consumer" class="com.aliyun.openservices.ons.api.bean.ConsumerBean" init-method="start" destroy-method="shutdown">
9.          <property name="properties" > <!--消费者配置信息-->
10.              <props>
11.                  <prop key="ConsumerId">CID_DEMO</prop> <!--请替换XXX-->
12.                  <prop key="AccessKey">AKDEMO</prop>
13.                  <prop key="SecretKey">SKDEMO</prop>
14.                  <!--将消费者线程数固定为50个
15.                  <prop key="ConsumeThreadNums">50</prop>
16.                  -->
17.              </props>
18.          </property>
19.          <property name="subscriptionTable">
20.              <map>
21.                  <entry value-ref="msgListener">
22.                      <key>
23.                          <bean class="com.aliyun.openservices.ons.api.bean.Subscription">
24.                              <property name="topic" value="TopicTestMQ"/>
25.                              <property name="expression" value="*"/><!--expression即Tag，可以设置成具体的Tag，如 taga||tagb||tagc，也可设置成*。 *仅代表订阅所有Tag，不支持通配-->
26.                          </bean>
27.                      </key>
28.                  </entry>
29.                  <!--更多的订阅添加entry节点即可，如下所示-->
30.                  <entry value-ref="msgListener">
31.                      <key>
32.                          <bean class="com.aliyun.openservices.ons.api.bean.Subscription">
33.                              <property name="topic" value="TopicTestMQ-Other"/> <!--订阅另外一个Topic -->
34.                              <property name="expression" value="taga||tagb"/> <!-- 订阅多个Tag -->
35.                          </bean>
36.                      </key>
37.                  </entry>
38.              </map>
39.          </property>
40.      </bean>
42. </beans>

1. package demo;
3. import org.springframework.context.ApplicationContext;
4. import org.springframework.context.support.ClassPathXmlApplicationContext;
6. public class ConsumeWithSpring {
7.      public static void main(String[] args) {
9.          /**
10.           * 消费者Bean配置在consumer.xml中，可通过ApplicationContext获取或者直接注入到其他类(比如具体的Controller)中
11.           */
12.          ApplicationContext context = new ClassPathXmlApplicationContext("consumer.xml");
13.          System.out.println("Consumer Started");
14.      }
15. }