huanghai
/
YltProject
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

'commit'

Browse Source
main
黄海 12 months ago
parent 5af9a3c12e
commit f3341faf00
3 changed files with 152 additions and 16 deletions
Show Stats Download Patch File Download Diff File

17
Ylt/ms-user/pom.xml
Unescape View File

@ -24,19 +24,22 @@
<artifactId>jfinal</artifactId>
<version>${jfinal.version}</version>
</dependency>
<!-- HikariCP -->
<!-- 数据库设计文档生成工具-->
<dependency>
<groupId>cn.smallbun.screw</groupId>
<artifactId>screw-core</artifactId>
<version>1.0.5</version>
</dependency>
<dependency>
<groupId>com.zaxxer</groupId>
<artifactId>HikariCP</artifactId>
<version>3.4.5</version>
<version>4.0.3</version>
</dependency>
<!-- 数据库设计文档生成工具-->
<dependency>
<groupId>cn.smallbun.screw</groupId>
<artifactId>screw-core</artifactId>
<version>1.0.5</version>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>4.8.0</version>
</dependency>
<dependency>

142
Ylt/ms-user/src/main/java/com/charge/tools/TestRocketMQ.java
Unescape View File

@ -0,0 +1,142 @@
package com.charge.tools;
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import java.io.IOException;
import java.util.List;
//Java整合RocketMQ实现生产消费
//https://blog.csdn.net/qq_28314431/article/details/128452607
public class TestRocketMQ {
private final static String nameServer = "10.10.14.210:9876";
private final static String producerGroup = "my_group";
private final static String consumerGroup = "my_group";
private final static String topic = "topic-test";
//同步发送
public static void syncSend() {
try {
// 初始化一个producer并设置Producer group name
DefaultMQProducer producer = new DefaultMQProducer(producerGroup);
// 设置NameServer地址
producer.setNamesrvAddr(nameServer);
// 启动producer
producer.start();
// 创建一条消息，并指定topic、tag、body等信息，tag可以理解成标签，对消息进行再归类，RocketMQ可以在消费端对tag进行过滤
Message msg = new Message(topic, "tagA", "Hello RocketMQ".getBytes(RemotingHelper.DEFAULT_CHARSET));
// 利用producer进行发送，并同步等待发送结果
SendResult sendResult = producer.send(msg, 10000);
System.out.printf("%s%n", sendResult);
// 一旦producer不再使用，关闭producer
producer.shutdown();
} catch (Exception e) {
e.printStackTrace();
}
}
//异步发送
public static void asyncSend() throws IOException {
try {
// 初始化一个producer并设置Producer group name
DefaultMQProducer producer = new DefaultMQProducer(producerGroup);
// 设置NameServer地址
producer.setNamesrvAddr(nameServer);
// 启动producer
producer.start();
// 创建一条消息，并指定topic、tag、body等信息，tag可以理解成标签，对消息进行再归类，RocketMQ可以在消费端对tag进行过滤
Message msg = new Message(topic, "tagB", "Hello RocketMQ".getBytes(RemotingHelper.DEFAULT_CHARSET));
// 异步发送消息, 发送结果通过callback返回给客户端
producer.send(msg, new SendCallback() {
public void onSuccess(SendResult sendResult) {
System.out.printf("OK %s %n",
sendResult.getMsgId());
}
public void onException(Throwable e) {
System.out.printf("Exception %s %n", e);
e.printStackTrace();
}
}, 10000);
} catch (Exception e) {
e.printStackTrace();
}
System.in.read();
}
//顺序消息
public static void orderSend() {
try {
// 初始化一个producer并设置Producer group name
DefaultMQProducer producer = new DefaultMQProducer(producerGroup);
// 设置NameServer地址
producer.setNamesrvAddr(nameServer);
producer.setSendMsgTimeout(10000);
// 启动producer
producer.start();
String[] tags = new String[]{"TagA", "TagB", "TagC"};
for (int i = 0; i < 10; i++) {
int orderId = i % 10;
Message msg = new Message(topic, tags[i % tags.length], "KEY" + i, ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
Integer id = (Integer) arg;
int index = id % mqs.size();
return mqs.get(index);
}
}, orderId);
System.out.printf("%s%n", sendResult);
}
} catch (Exception e) {
e.printStackTrace();
}
}
public static void consumerPull() {
try {
DefaultLitePullConsumer consumer = new DefaultLitePullConsumer(consumerGroup);
consumer.setNamesrvAddr(nameServer);
//关闭自动提交
consumer.setAutoCommit(false);
consumer.subscribe(topic, "*");
consumer.setPullBatchSize(20);
consumer.start();
while (true) {
List<MessageExt> messageExts = consumer.poll();
if (messageExts.size() > 0) {
String str = new String(messageExts.get(0).getBody());
System.out.println(str);
} else {
System.out.println("本轮没有找到需要消费的消息！");
}
consumer.commitSync();
}
} catch (Exception e) {
e.printStackTrace();
}
}
public static void main(String[] args) throws IOException {
//同步发送
syncSend();
//异步发送
//asyncSend();
//顺序消息
//orderSend();
//拉消费
consumerPull();
}
}

9
Ylt/ms-user/src/test/java/UserTest.java
Unescape View File

@ -2,14 +2,12 @@ import com.charge.UserApplication;
import com.charge.rpc.facade.UserFacade;
import com.jfinal.plugin.activerecord.Db;
import com.jfinal.plugin.activerecord.Record;
import com.jfinal.plugin.activerecord.SqlPara;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
import javax.annotation.Resource;
import java.util.List;
@RunWith(SpringRunner.class)
@ -33,11 +31,4 @@ public class UserTest {
System.out.println(record);
}
@Test
public void getAll() {
SqlPara sqlPara = Db.getSqlPara("Test.selectAll");
List<Record> list = Db.find(sqlPara);
System.out.println(list);
}
}

Write Preview
Loading…
Cancel
Save