msgframe的rocketmq测试代码

ipu-rocketmq-example/src/main/java/com/ai/ipu/example/rocketmq/RocketMQCustomer.java

@@ -0,0 +1,15 @@
package com.ai.ipu.example.rocketmq;
import com.ai.aif.msgframe.consumer.MfServiceStartup;
/**
 * @author liutong3
 * @team IPU
 * @date 2019/6/10 14:13
 * @desc 启动RocketMQ消费者
 */
public class RocketMQCustomer {
    public static void main(String[] args){
        MfServiceStartup.main();
    }
}

ipu-rocketmq-example/src/main/java/com/ai/ipu/example/rocketmq/RocketMQCustomerExample.java

@@ -0,0 +1,23 @@
package com.ai.ipu.example.rocketmq;
import com.ai.aif.msgframe.common.IConsumerProcessor;
import com.ai.aif.msgframe.common.exception.ConsumerException;
import com.ai.aif.msgframe.common.message.MsgFMessage;
import com.ai.aif.msgframe.common.message.MsgFTextMessage;
/**
 * @author liutong3
 * @team IPU
 * @date 2019/6/10 14:15
 * @desc RocketMQ消费者
 */
public class RocketMQCustomerExample implements IConsumerProcessor {
    @SuppressWarnings("rawtypes")
    public Object process(MsgFMessage message) throws ConsumerException {
        if (message instanceof MsgFTextMessage) {
            System.out.println("{topic:" + message.getTopic() + ",value:" + ((MsgFTextMessage) message).getText() + "}");
        }
        return true;
    }
}

ipu-rocketmq-example/src/main/java/com/ai/ipu/example/rocketmq/RocketMQProducerExample.java

@@ -0,0 +1,114 @@
package com.ai.ipu.example.rocketmq;
import com.ai.aif.msgframe.MfProducerClient;
import com.ai.aif.msgframe.common.CompletionListener;
import com.ai.aif.msgframe.common.message.MsgFMessage;
import com.ai.aif.msgframe.common.message.MsgFTextMessage;
import org.apache.log4j.Logger;
import org.junit.Test;
import java.util.Date;
/**
 * @author liutong3
 * @team IPU
 * @date 2019/6/10 14:16
 * @desc RocketMQ生产者
 */
public class RocketMQProducerExample {
    private static final Logger LOG = Logger.getLogger(RocketMQProducerExample.class);
    private final String TOPIC = "test";
    private final String MESSAGE = "test";
    private final String ORDERID = "1000";
    @Test
    public void testSendMessage() {
        String message = getMessage();
        sendMessage(TOPIC, message);
    }
    @Test
    public void testSendAsyncMessage() {
        String message = getMessage();
        sendAsyncMessage(TOPIC, message);
    }
    @Test
    public void testSendOrderMessage() {
        String message = getMessage();
        sendOrderMessage(TOPIC, message, ORDERID);
    }
    @Test
    public void testSendOneWay() {
        String message = getMessage();
        sendOneWay(TOPIC, message);
    }
    private String getMessage() {
        return MESSAGE + "_" + new Date().getTime();
    }
    private void sendMessage(String topic, String text) {
        try {
            MfProducerClient client = new MfProducerClient();
            MsgFTextMessage message = new MsgFTextMessage();
            message.setText(text);
            client.send(topic, message);
            LOG.debug("Sent:" + message);
            Thread.currentThread().sleep(1000);
        } catch (Exception e) {
            LOG.debug(e);
            throw new RuntimeException(e);
        }
    }
    private void sendAsyncMessage(String topic, final String text) {
        try {
            MfProducerClient client = new MfProducerClient();
            final MsgFTextMessage message = new MsgFTextMessage();
            message.setText(text);
            client.asyncSend(topic, message, new CompletionListener() {
                public void onCompletion(MsgFMessage msgFMessage) {
                    LOG.debug("Sent Success:" + msgFMessage.getMsgId() + "-" + text);
                }
                public void onException(MsgFMessage msgFMessage, Exception e) {
                    LOG.debug("Sent Error:" + msgFMessage.getMsgId() + "-" + text);
                }
            });
            Thread.currentThread().sleep(1000);
        } catch (Exception e) {
            LOG.debug(e);
            throw new RuntimeException(e);
        }
    }
    private void sendOrderMessage(String topic, final String text, String orderId) {
        try {
            MfProducerClient client = new MfProducerClient();
            final MsgFTextMessage message = new MsgFTextMessage();
            message.setText(text);
            client.sendOrderMsg(topic, message, orderId);
            LOG.debug("Sent:" + message);
            Thread.currentThread().sleep(1000);
        } catch (Exception e) {
            LOG.debug(e);
            throw new RuntimeException(e);
        }
    }
    private void sendOneWay(String topic, final String text) {
        try {
            MfProducerClient client = new MfProducerClient();
            final MsgFTextMessage message = new MsgFTextMessage();
            message.setText(text);
            client.sendOneway(topic, message);
            LOG.debug("Sent:" + message);
            Thread.currentThread().sleep(1000);
        } catch (Exception e) {
            LOG.debug(e);
            throw new RuntimeException(e);
        }
    }
}

ipu-rocketmq-example/src/main/resources/log4j.properties

@@ -0,0 +1,8 @@
log4j.rootLogger=error, console
log4j.logger.org.apache.zookeeper=error
log4j.logger.com.ai.ipu.example=debug
log4j.appender.console=org.apache.log4j.ConsoleAppender
log4j.appender.console.target=System.out
log4j.appender.console.layout=org.apache.log4j.PatternLayout
log4j.appender.console.layout.ConversionPattern=%d{yy/MM/dd HH:mm:ss} %5p [%t] (%F:%L) - %m%n

ipu-rocketmq-example/src/main/resources/msgframe-config.xml

@@ -0,0 +1,22 @@
<?xml version="1.0" encoding="UTF-8"?>
<msgframeCfg xmlns="http://www.asiainfo.com/msgframe">
    <centerCfg>
        <name>ipu-rocketmq-example</name>
        <destinations>
            <queue name="test" belong="myCenter"/>
        </destinations>
        <subscribes>
            <subscribe subDestination="test">
                <implclass>com.ai.ipu.example.rocketmq.RocketMQCustomerExample</implclass>
            </subscribe>
        </subscribes>
        <centers>
            <center name="myCenter" containClusters="rocketmq-cluster"/>
        </centers>
        <clusters>
            <cluster name="rocketmq-cluster" type="RocketMQ">
                <url>47.105.160.21:9876</url>
            </cluster>
        </clusters>
    </centerCfg>
</msgframeCfg>