Spring 集成__Java SDK_TCP 接入(專業)_消息隊列 MQ-阿裏雲

2.通過已經與 Spring 集成好的生產者生產消息。

package demo;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.Producer;
import com.aliyun.openservices.ons.api.SendResult;
import com.aliyun.openservices.ons.api.exception.ONSClientException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
public class ProduceWithSpring {
    public static void main(String[] args) {
        /**
         * 生產者Bean配置在producer.xml中,可通過ApplicationContext獲取或者直接注入到其他類(比如具體的Controller)中.
         */
        ApplicationContext context = new ClassPathXmlApplicationContext("producer.xml");
        Producer producer = (Producer) context.getBean("producer");
        //循環發送消息
        for (int i = 0; i < 100; i++) {
            Message msg = new Message( //
                    // Message Topic
                    "TopicTestMQ",
                    // Message Tag 可理解為Gmail中的標簽，對消息進行再歸類，方便Consumer指定過濾條件在MQ服務器過濾
                    "TagA",
                    // Message Body 可以是任何二進製形式的數據， MQ不做任何幹預
                    // 需要Producer與Consumer協商好一致的序列化和反序列化方式
                    "Hello MQ".getBytes());
            // 設置代表消息的業務關鍵屬性，請盡可能全局唯一
            // 以方便您在無法正常收到消息情況下，可通過MQ 控製台查詢消息並補發
            // 注意：不設置也不會影響消息正常收發
            msg.setKey("ORDERID_100");
            // 發送消息，隻要不拋異常就是成功
            try {
                SendResult sendResult = producer.send(msg);
                assert sendResult != null;
                System.out.println("send success: " + sendResult.getMessageId());
            }catch (ONSClientException e) {
                System.out.println("發送失敗");
            }
        }
    }
}

package demo;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.transaction.LocalTransactionChecker;
import com.aliyun.openservices.ons.api.transaction.TransactionStatus;
public class DemoLocalTransactionChecker implements LocalTransactionChecker {
    public TransactionStatus check(Message msg) {
        System.out.println("開始回查本地事務狀態");
        return TransactionStatus.CommitTransaction; //根據本地事務狀態檢查結果返回不同的TransactionStatus
    }
}

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="https://www.springframework.org/schema/beans"
       xmlns:xsi="https://www.w3.org/2001/XMLSchema-instance"
       xsi:schemaLocation="https://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd">
    <bean id="localTransactionChecker" class="demo.DemoLocalTransactionChecker"></bean>
    <bean id="transactionProducer" class="com.aliyun.openservices.ons.api.bean.TransactionProducerBean" init-method="start" destroy-method="shutdown">
        <property name="properties" > <!--事務消息生產者配置信息-->
            <props>
                <prop key="ProducerId">PID_DEMO</prop> <!--請替換XXX-->
                <prop key="AccessKey">AKDEMO</prop>
                <prop key="SecretKey">SKDEMO</prop>
            </props>
        </property>
        <property name="localTransactionChecker" ref="localTransactionChecker"></property>
    </bean>
</beans>

package demo;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.SendResult;
import com.aliyun.openservices.ons.api.transaction.LocalTransactionExecuter;
import com.aliyun.openservices.ons.api.transaction.TransactionProducer;
import com.aliyun.openservices.ons.api.transaction.TransactionStatus;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
public class ProduceTransMsgWithSpring {
    public static void main(String[] args) {
        /**
         * 事務消息生產者Bean配置在transactionProducer.xml中,可通過ApplicationContext獲取或者直接注入到其他類(比如具體的Controller)中.
         * 請結合例子"發送事務消息"
         */
        ApplicationContext context = new ClassPathXmlApplicationContext("transactionProducer.xml");
        TransactionProducer transactionProducer = (TransactionProducer) context.getBean("transactionProducer");
        Message msg = new Message("XXX", "TagA", "Hello MQ transaction===".getBytes());
        SendResult sendResult = transactionProducer.send(msg, new LocalTransactionExecuter() {
            @Override
            public TransactionStatus execute(Message msg, Object arg) {
                System.out.println("執行本地事務");
                return TransactionStatus.CommitTransaction; //根據本地事務執行結果來返回不同的TransactionStatus
            }
        }, null);
    }
}

package demo;
import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.MessageListener;
public class DemoMessageListener implements MessageListener {
    public Action consume(Message message, ConsumeContext context) {
        System.out.println("Receive: " + message.getMsgID());
        try {
            //do something..
            return Action.CommitMessage;
        }catch (Exception e) {
            //消費失敗
            return Action.ReconsumeLater;
        }
    }
}

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="https://www.springframework.org/schema/beans"
       xmlns:xsi="https://www.w3.org/2001/XMLSchema-instance"
       xsi:schemaLocation="https://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd">
    <bean id="msgListener" class="demo.DemoMessageListener"></bean> <!--Listener配置-->
    <bean id="consumer" class="com.aliyun.openservices.ons.api.bean.ConsumerBean" init-method="start" destroy-method="shutdown">
        <property name="properties" > <!--消費者配置信息-->
            <props>
                <prop key="ConsumerId">CID_DEMO</prop> <!--請替換XXX-->
                <prop key="AccessKey">AKDEMO</prop>
                <prop key="SecretKey">SKDEMO</prop>
                <!--將消費者線程數固定為50個
                <prop key="ConsumeThreadNums">50</prop>
                -->
            </props>
        </property>
        <property name="subscriptionTable">
            <map>
                <entry value-ref="msgListener">
                    <key>
                        <bean class="com.aliyun.openservices.ons.api.bean.Subscription">
                            <property name="topic" value="TopicTestMQ"/>
                            <property name="expression" value="*"/><!--expression即Tag，可以設置成具體的Tag，如 taga||tagb||tagc，也可設置成*。 *僅代表訂閱所有Tag，不支持通配-->
                        </bean>
                    </key>
                </entry>
                <!--更多的訂閱添加entry節點即可-->
            </map>
        </property>
    </bean>
</beans>

package demo;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
public class ConsumeWithSpring {
    public static void main(String[] args) {
        /**
         * 消費者Bean配置在consumer.xml中，可通過ApplicationContext獲取或者直接注入到其他類(比如具體的Controller)中
         */
        ApplicationContext context = new ClassPathXmlApplicationContext("consumer.xml");
        System.out.println("Consumer Started");
    }
}