msgframe的kafka测试代码

ipu-kafka-example/ipu-kafka-example.iml

@@ -0,0 +1,37 @@
<?xml version="1.0" encoding="UTF-8"?>
<module org.jetbrains.idea.maven.project.MavenProjectsManager.isMavenModule="true" type="JAVA_MODULE" version="4">
  <component name="NewModuleRootManager" LANGUAGE_LEVEL="JDK_1_5">
    <output url="file://$MODULE_DIR$/target/classes" />
    <output-test url="file://$MODULE_DIR$/target/test-classes" />
    <content url="file://$MODULE_DIR$">
      <sourceFolder url="file://$MODULE_DIR$/src/main/java" isTestSource="false" />
      <sourceFolder url="file://$MODULE_DIR$/src/main/resources" type="java-resource" />
      <sourceFolder url="file://$MODULE_DIR$/src/test/java" isTestSource="true" />
      <excludeFolder url="file://$MODULE_DIR$/target" />
    </content>
    <orderEntry type="inheritedJdk" />
    <orderEntry type="sourceFolder" forTests="false" />
    <orderEntry type="library" name="Maven: com.ai.aif.msgframe:msgframe-common:1.9.2" level="project" />
    <orderEntry type="library" name="Maven: com.ai.aif.amber:amber-client:1.1.3" level="project" />
    <orderEntry type="library" name="Maven: org.fusesource.hawtbuf:hawtbuf:1.9" level="project" />
    <orderEntry type="library" name="Maven: com.aliyun.openservices:ons-client:1.7.0.Final" level="project" />
    <orderEntry type="library" name="Maven: commons-io:commons-io:2.4" level="project" />
    <orderEntry type="library" name="Maven: commons-lang:commons-lang:2.6" level="project" />
    <orderEntry type="library" name="Maven: com.ai.aif.msgframe:msgframe-client:1.9.2" level="project" />
    <orderEntry type="library" name="Maven: com.ai.aif.msgframe:msgframe-server:1.9.2" level="project" />
    <orderEntry type="library" name="Maven: com.ai.aif.msgframe:xmlmsgframe:2.6.4" level="project" />
    <orderEntry type="library" name="Maven: org.apache.xmlbeans:xmlbeans:2.6.0" level="project" />
    <orderEntry type="library" name="Maven: stax:stax-api:1.0.1" level="project" />
    <orderEntry type="library" name="Maven: com.rabbitmq:amqp-client:3.6.5" level="project" />
    <orderEntry type="library" name="Maven: com.alibaba:fastjson:1.1.46" level="project" />
    <orderEntry type="library" name="Maven: org.hamcrest:hamcrest-core:1.3" level="project" />
    <orderEntry type="library" name="Maven: javax.jms:jms:1.1" level="project" />
    <orderEntry type="library" name="Maven: org.apache.kafka:kafka-clients:0.10.2.0" level="project" />
    <orderEntry type="library" name="Maven: net.jpountz.lz4:lz4:1.3.0" level="project" />
    <orderEntry type="library" name="Maven: org.xerial.snappy:snappy-java:1.1.2.6" level="project" />
    <orderEntry type="library" name="Maven: org.slf4j:slf4j-api:1.7.21" level="project" />
    <orderEntry type="library" name="Maven: log4j:log4j:1.2.12" level="project" />
    <orderEntry type="library" name="Maven: com.alibaba.rocketmq:rocketmq-client:3.5.9" level="project" />
    <orderEntry type="library" name="Maven: junit:junit:4.12" level="project" />
  </component>
</module>

ipu-kafka-example/pom.xml

@@ -0,0 +1,86 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>com.ai.ipu</groupId>
    <artifactId>ipu-kafka-example</artifactId>
    <version>1.0-SNAPSHOT</version>
    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <ipu>3.1-SNAPSHOT</ipu>
        <jdk>1.8</jdk>
    </properties>
    <dependencies>
        <dependency>
            <groupId>com.ai.aif.msgframe</groupId>
            <artifactId>msgframe-common</artifactId>
            <version>1.9.2</version>
        </dependency>
        <dependency>
            <groupId>com.ai.aif.msgframe</groupId>
            <artifactId>msgframe-client</artifactId>
            <version>1.9.2</version>
        </dependency>
        <dependency>
            <groupId>com.ai.aif.msgframe</groupId>
            <artifactId>msgframe-server</artifactId>
            <version>1.9.2</version>
        </dependency>
        <dependency>
            <groupId>com.ai.aif.msgframe</groupId>
            <artifactId>xmlmsgframe</artifactId>
            <version>2.6.4</version>
        </dependency>
        <dependency>
            <groupId>org.apache.xmlbeans</groupId>
            <artifactId>xmlbeans</artifactId>
            <version>2.6.0</version>
        </dependency>
        <dependency>
            <groupId>com.rabbitmq</groupId>
            <artifactId>amqp-client</artifactId>
            <version>3.6.5</version>
        </dependency>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.1.46</version>
        </dependency>
        <dependency>
            <groupId>org.hamcrest</groupId>
            <artifactId>hamcrest-core</artifactId>
            <version>1.3</version>
        </dependency>
        <dependency>
            <groupId>javax.jms</groupId>
            <artifactId>jms</artifactId>
            <version>1.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>0.10.2.0</version>
        </dependency>
        <dependency>
            <groupId>log4j</groupId>
            <artifactId>log4j</artifactId>
            <version>1.2.12</version>
        </dependency>
        <dependency>
            <groupId>com.alibaba.rocketmq</groupId>
            <artifactId>rocketmq-client</artifactId>
            <version>3.5.9</version>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.12</version>
            <scope>compile</scope>
        </dependency>
    </dependencies>
</project>

ipu-kafka-example/src/main/java/com/ai/ipu/example/kafka/KafkaCustomer.java

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

ipu-kafka-example/src/main/java/com/ai/ipu/example/kafka/KafkaCustomerExample.java

@@ -0,0 +1,22 @@
package com.ai.ipu.example.kafka;
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/6 14:59
 * @desc kafka消费者
 */
public class KafkaCustomerExample 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-kafka-example/src/main/java/com/ai/ipu/example/kafka/KafkaProducerExample.java

@@ -0,0 +1,91 @@
package com.ai.ipu.example.kafka;
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/6 16:07
 * @desc kafka生产者
 */
public class KafkaProducerExample {
    private static final Logger LOG = Logger.getLogger(KafkaProducerExample.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);
    }
    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) {
            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) {
            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);
            Thread.currentThread().sleep(1000);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}

ipu-kafka-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-kafka-example/src/main/resources/msgframe-config.xml

@@ -0,0 +1,28 @@
<?xml version="1.0" encoding="UTF-8"?>
<msgframeCfg xmlns="http://www.asiainfo.com/msgframe">
    <centerCfg>
        <name>ipu-kafka-example</name>
        <destinations>
            <queue name="test" belong="myCenter"/>
        </destinations>
        <subscribes>
            <subscribe subDestination="test">
                <implclass>com.ai.ipu.example.kafka.KafkaCustomerExample</implclass>
            </subscribe>
        </subscribes>
        <centers>
            <center name="myCenter" containClusters="kafka-cluster"/>
        </centers>
        <clusters>
            <cluster name="kafka-cluster" type="Kafka">
                <url>47.105.160.21:9091,47.105.160.21:9092,47.105.160.21:9093</url>
            </cluster>
        </clusters>
        <!--<persistence>-->
            <!--<exceptionPersistence>-->
                <!--<producerExceptionClass>com.ai.ipu.example.kafka.exception.ProducerExceprionHandle</producerExceptionClass>-->
                <!--<consumerExceptionClass>com.ai.ipu.example.kafka.exception.ConsumerExceprionHandle</consumerExceptionClass>-->
            <!--</exceptionPersistence>-->
        <!--</persistence>-->
    </centerCfg>
</msgframeCfg>

ipu-zk-example/ipu-zk-example.iml

@@ -11,8 +11,6 @@
    </content>
    <orderEntry type="inheritedJdk" />
    <orderEntry type="sourceFolder" forTests="false" />
    <orderEntry type="library" name="Maven: junit:junit:4.12" level="project" />
    <orderEntry type="library" name="Maven: org.hamcrest:hamcrest-core:1.3" level="project" />
    <orderEntry type="library" name="Maven: com.ai.ipu:ipu-zk:3.1-SNAPSHOT" level="project" />
    <orderEntry type="library" name="Maven: org.apache.zookeeper:zookeeper:3.4.9" level="project" />
    <orderEntry type="library" name="Maven: org.slf4j:slf4j-api:1.6.1" level="project" />
@@ -51,5 +49,7 @@
    <orderEntry type="library" name="Maven: com.alibaba:fastjson:1.2.38" level="project" />
    <orderEntry type="library" name="Maven: org.yaml:snakeyaml:1.23" level="project" />
    <orderEntry type="library" name="Maven: com.ai.wade:wade-data:1.0" level="project" />
    <orderEntry type="library" name="Maven: junit:junit:4.12" level="project" />
    <orderEntry type="library" name="Maven: org.hamcrest:hamcrest-core:1.3" level="project" />
  </component>
</module>