whatsmars-mq rocketmq

pom.xml

@@ -25,6 +25,7 @@
         <module>whatsmars-worker</module>
         <module>whatsmars-rpc</module>
         <module>whatsmars-dubbo</module>
+        <module>whatsmars-mq</module>
     </modules>
 
     <properties>

whatsmars-mq/pom.xml

+<?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">
+    <parent>
+        <artifactId>whatsmars-parent</artifactId>
+        <groupId>com.itlong</groupId>
+        <version>1.0-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>whatsmars-mq</artifactId>
+
+    <dependencies>
+        <!-- https://mvnrepository.com/artifact/org.apache.rocketmq/rocketmq-namesrv -->
+        <dependency>
+            <groupId>org.apache.rocketmq</groupId>
+            <artifactId>rocketmq-namesrv</artifactId>
+            <version>4.0.0-incubating</version>
+        </dependency>
+        <!-- https://mvnrepository.com/artifact/org.apache.rocketmq/rocketmq-broker -->
+        <dependency>
+            <groupId>org.apache.rocketmq</groupId>
+            <artifactId>rocketmq-broker</artifactId>
+            <version>4.0.0-incubating</version>
+        </dependency>
+        <!-- https://mvnrepository.com/artifact/org.apache.rocketmq/rocketmq-client -->
+        <dependency>
+            <groupId>org.apache.rocketmq</groupId>
+            <artifactId>rocketmq-client</artifactId>
+            <version>4.0.0-incubating</version>
+        </dependency>
+
+    </dependencies>
+
+
+</project>
\ No newline at end of file

whatsmars-mq/src/main/java/com/itlong/whatsmars/mq/rocketmq/quickstart/BrokerStartup.java

+package com.itlong.whatsmars.mq.rocketmq.quickstart;
+
+/**
+ * Created by shenhongxi on 2017/6/21.
+ */
+public class BrokerStartup {
+
+    public static void main(String[] args) {
+        args = new String[] {"-c", "D:\\github\\whatsmars\\whatsmars-mq\\src\\main\\resources\\conf.properties"};
+        org.apache.rocketmq.broker.BrokerStartup.main(args);
+        System.out.println("Broker started.");
+    }
+}

whatsmars-mq/src/main/java/com/itlong/whatsmars/mq/rocketmq/quickstart/Consumer.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.itlong.whatsmars.mq.rocketmq.quickstart;
+
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
+import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
+import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
+import org.apache.rocketmq.common.message.MessageExt;
+
+import java.util.List;
+
+/**
+ * This example shows how to subscribe and consume messages using providing {@link DefaultMQPushConsumer}.
+ */
+public class Consumer {
+
+    public static void main(String[] args) throws InterruptedException, MQClientException {
+
+        /*
+         * Instantiate with specified consumer group name.
+         */
+        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_4");
+
+        /*
+         * Specify name server addresses.
+         * <p/>
+         *
+         * Alternatively, you may specify name server addresses via exporting environmental variable: NAMESRV_ADDR
+         * <pre>
+         * {@code
+         * consumer.setNamesrvAddr("name-server1-ip:9876;name-server2-ip:9876");
+         * }
+         * </pre>
+         */
+        consumer.setNamesrvAddr("127.0.0.1:9876");
+        /*
+         * Specify where to start in case the specified consumer group is a brand new one.
+         */
+        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
+
+        /*
+         * Subscribe one more more topics to consume.
+         */
+        consumer.subscribe("TopicTest", "*");
+
+        /*
+         *  Register callback to execute on arrival of messages fetched from brokers.
+         */
+        consumer.registerMessageListener(new MessageListenerConcurrently() {
+
+            @Override
+            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
+                                                            ConsumeConcurrentlyContext context) {
+                System.out.printf(Thread.currentThread().getName() + " Receive New Messages: " + msgs + "%n");
+                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
+            }
+        });
+
+        /*
+         *  Launch the consumer instance.
+         */
+        consumer.start();
+
+        System.out.printf("Consumer Started.%n");
+    }
+}

whatsmars-mq/src/main/java/com/itlong/whatsmars/mq/rocketmq/quickstart/NamesrvStartup.java

+package com.itlong.whatsmars.mq.rocketmq.quickstart;
+
+/**
+ * Created by shenhongxi on 2017/6/21.
+ */
+public class NamesrvStartup {
+
+    public static void main(String[] args) {
+        args = new String[] {"-c", "D:\\github\\whatsmars\\whatsmars-mq\\src\\main\\resources\\conf.properties"};
+        org.apache.rocketmq.namesrv.NamesrvStartup.main(args);
+    }
+}

whatsmars-mq/src/main/java/com/itlong/whatsmars/mq/rocketmq/quickstart/Producer.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.itlong.whatsmars.mq.rocketmq.quickstart;
+
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.remoting.common.RemotingHelper;
+
+/**
+ * This class demonstrates how to send messages to brokers using provided {@link DefaultMQProducer}.
+ */
+public class Producer {
+    public static void main(String[] args) throws MQClientException, InterruptedException {
+
+        /*
+         * Instantiate with a producer group name.
+         */
+        DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
+
+        /*
+         * Specify name server addresses.
+         * <p/>
+         *
+         * Alternatively, you may specify name server addresses via exporting environmental variable: NAMESRV_ADDR
+         * <pre>
+         * {@code
+         * producer.setNamesrvAddr("name-server1-ip:9876;name-server2-ip:9876");
+         * }
+         * </pre>
+         */
+        producer.setNamesrvAddr("127.0.0.1:9876");
+
+        /*
+         * Launch the instance.
+         */
+        producer.start();
+
+        for (int i = 0; i < 1000; i++) {
+            try {
+
+                /*
+                 * Create a message instance, specifying topic, tag and message body.
+                 */
+                Message msg = new Message("TopicTest" /* Topic */,
+                    "TagA" /* Tag */,
+                    ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
+                );
+
+                /*
+                 * Call send message to deliver message to one of brokers.
+                 */
+                SendResult sendResult = producer.send(msg);
+
+                System.out.printf("%s%n", sendResult);
+            } catch (Exception e) {
+                e.printStackTrace();
+                Thread.sleep(1000);
+            }
+        }
+
+        /*
+         * Shut down once the producer instance is not longer in use.
+         */
+        producer.shutdown();
+    }
+}

whatsmars-mq/src/main/resources/conf.properties

+rocketmqHome=D:\\github\\incubator-rocketmq\\distribution
+namesrvAddr=127.0.0.1:9876
+mapedFileSizeCommitLog=52428800
+mapedFileSizeConsumeQueue=30000
\ No newline at end of file