Skip to content

Commit 758cc2f

Browse files
hengboyhengboy
committed
✅ 添加v1.0.3版本消息管道新特性示例
1 parent 1c1b60e commit 758cc2f

File tree

7 files changed

+118
-3
lines changed

7 files changed

+118
-3
lines changed

‎api-boot-samples/api-boot-sample-message-pipe-client/pom.xml‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,11 @@
2020
<artifactId>redisson-spring-boot-starter</artifactId>
2121
<version>3.13.3</version>
2222
</dependency>
23+
<dependency>
24+
<groupId>com.alibaba.nacos</groupId>
25+
<artifactId>nacos-client</artifactId>
26+
<version>1.3.2</version>
27+
</dependency>
2328
<dependency>
2429
<groupId>org.minbox.framework</groupId>
2530
<artifactId>api-boot-starter-message-pipe-client</artifactId>

‎api-boot-samples/api-boot-sample-message-pipe-client/src/main/java/org/minbox/framework/api/boot/sample/message/pipe/client/MessagePipeClientConfiguration.java‎

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,38 @@
1+
package org.minbox.framework.api.boot.sample.message.pipe.client;
2+
3+
import com.alibaba.nacos.api.NacosFactory;
4+
import com.alibaba.nacos.api.PropertyKeyConst;
5+
import com.alibaba.nacos.api.exception.NacosException;
6+
import com.alibaba.nacos.api.naming.NamingService;
7+
import org.minbox.framework.message.pipe.spring.annotation.ServerServiceType;
8+
import org.minbox.framework.message.pipe.spring.annotation.client.EnableMessagePipeClient;
9+
import org.springframework.context.annotation.Bean;
10+
import org.springframework.context.annotation.Configuration;
11+
12+
import java.util.Properties;
13+
14+
/**
15+
* 消息管道客户端配置示例
16+
* <p>
17+
* 使用Nacos方式进行注册服务
18+
*
19+
* @author 恒宇少年
20+
*/
21+
@Configuration
22+
@EnableMessagePipeClient(serverType = ServerServiceType.NACOS)
23+
public class MessagePipeClientConfiguration {
24+
/**
25+
* 配置Nacos {@link NamingService}实例
26+
*
27+
* @return
28+
* @throws NacosException
29+
*/
30+
@Bean
31+
public NamingService namingService() throws NacosException {
32+
Properties properties = new Properties();
33+
properties.put(PropertyKeyConst.USERNAME, "nacos");
34+
properties.put(PropertyKeyConst.PASSWORD, "nacos");
35+
properties.put(PropertyKeyConst.SERVER_ADDR, "open.nacos.yuqiyu.com:80");
36+
return NacosFactory.createNamingService(properties);
37+
}
38+
}

‎api-boot-samples/api-boot-sample-message-pipe-client/src/main/java/org/minbox/framework/api/boot/sample/message/pipe/client/MessagePipeClientSampleApplication.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,7 @@
11
package org.minbox.framework.api.boot.sample.message.pipe.client;
22

3+
import org.minbox.framework.message.pipe.spring.annotation.ServerServiceType;
4+
import org.minbox.framework.message.pipe.spring.annotation.client.EnableMessagePipeClient;
35
import org.slf4j.Logger;
46
import org.slf4j.LoggerFactory;
57
import org.springframework.boot.SpringApplication;

‎api-boot-samples/api-boot-sample-message-pipe-client/src/main/java/org/minbox/framework/api/boot/sample/message/pipe/client/processor/TestMessageProcessor.java‎

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2,8 +2,12 @@
22

33
import lombok.extern.slf4j.Slf4j;
44
import org.minbox.framework.message.pipe.client.process.MessageProcessor;
5+
import org.minbox.framework.message.pipe.core.Message;
6+
import org.minbox.framework.message.pipe.core.untis.JsonUtils;
57
import org.springframework.stereotype.Service;
68

9+
import java.util.Map;
10+
711
/**
812
* 示例 {@link MessageProcessor}
913
*
@@ -20,8 +24,11 @@ public String bindingPipeName() {
2024
}
2125

2226
@Override
23-
public boolean processing(String specificPipeName, String requestId, byte[] messageBody) {
24-
log.info("具体管道：{}，消费消息：{}，内容：{}", specificPipeName, requestId, new String(messageBody));
27+
public boolean processing(String specificPipeName, String requestId, Message message) {
28+
Map<String, Object> metadata = message.getMetadata();
29+
byte[] messageBody = message.getBody();
30+
log.info("具体管道：{}，消费消息：{}，内容：{}，元数据：{}",
31+
specificPipeName, requestId, new String(messageBody), JsonUtils.objectToJson(metadata));
2532
return true;
2633
}
2734
}

‎api-boot-samples/api-boot-sample-message-pipe-server/pom.xml‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,11 @@
2020
<artifactId>redisson-spring-boot-starter</artifactId>
2121
<version>3.13.3</version>
2222
</dependency>
23+
<dependency>
24+
<groupId>com.alibaba.nacos</groupId>
25+
<artifactId>nacos-client</artifactId>
26+
<version>1.3.2</version>
27+
</dependency>
2328
<dependency>
2429
<groupId>org.minbox.framework</groupId>
2530
<artifactId>api-boot-starter-message-pipe-server</artifactId>

‎api-boot-samples/api-boot-sample-message-pipe-server/src/main/java/org/minbox/framework/api/boot/sample/message/pipe/server/MessagePipeServerConfiguration.java‎

Lines changed: 21 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,9 @@
11
package org.minbox.framework.api.boot.sample.message.pipe.server;
22

3+
import com.alibaba.nacos.api.NacosFactory;
4+
import com.alibaba.nacos.api.PropertyKeyConst;
5+
import com.alibaba.nacos.api.exception.NacosException;
6+
import com.alibaba.nacos.api.naming.NamingService;
37
import org.minbox.framework.message.pipe.server.processing.push.PushMessageEvent;
48
import org.minbox.framework.message.pipe.spring.annotation.ServerServiceType;
59
import org.minbox.framework.message.pipe.spring.annotation.server.EnableMessagePipeServer;
@@ -8,13 +12,15 @@
812
import org.springframework.data.redis.connection.RedisConnectionFactory;
913
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
1014

15+
import java.util.Properties;
16+
1117
/**
1218
* 消息管道Server相关配置
1319
*
1420
* @author 恒宇少年
1521
*/
1622
@Configuration
17-
@EnableMessagePipeServer(serverType = ServerServiceType.GRPC)
23+
@EnableMessagePipeServer(serverType = ServerServiceType.NACOS)
1824
public class MessagePipeServerConfiguration {
1925
/**
2026
* 配置Redis监听容器
@@ -32,4 +38,18 @@ public RedisMessageListenerContainer redisMessageListenerContainer(RedisConnecti
3238
return container;
3339
}
3440

41+
/**
42+
* 配置Nacos {@link NamingService}实例
43+
*
44+
* @return
45+
* @throws NacosException
46+
*/
47+
@Bean
48+
public NamingService namingService() throws NacosException {
49+
Properties properties = new Properties();
50+
properties.put(PropertyKeyConst.USERNAME, "nacos");
51+
properties.put(PropertyKeyConst.PASSWORD, "nacos");
52+
properties.put(PropertyKeyConst.SERVER_ADDR, "open.nacos.yuqiyu.com:80");
53+
return NacosFactory.createNamingService(properties);
54+
}
3555
}

‎api-boot-samples/api-boot-sample-message-pipe-server/src/main/java/org/minbox/framework/api/boot/sample/message/pipe/server/PutMessage.java‎

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,38 @@
1+
package org.minbox.framework.api.boot.sample.message.pipe.server;
2+
3+
import org.minbox.framework.message.pipe.core.Message;
4+
import org.minbox.framework.message.pipe.server.MessagePipe;
5+
import org.minbox.framework.message.pipe.server.manager.MessagePipeManager;
6+
import org.springframework.beans.factory.InitializingBean;
7+
import org.springframework.beans.factory.annotation.Autowired;
8+
import org.springframework.context.annotation.Configuration;
9+
10+
import java.util.UUID;
11+
import java.util.concurrent.Executors;
12+
import java.util.concurrent.ScheduledExecutorService;
13+
import java.util.concurrent.TimeUnit;
14+
15+
/**
16+
* 将消息{@link Message}定时写入消息管道{@link MessagePipe}测试类
17+
*
18+
* @author 恒宇少年
19+
*/
20+
@Configuration
21+
public class PutMessage implements InitializingBean {
22+
@Autowired
23+
private MessagePipeManager messagePipeManager;
24+
private ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
25+
26+
@Override
27+
public void afterPropertiesSet() throws Exception {
28+
scheduledExecutorService.scheduleWithFixedDelay(() -> {
29+
MessagePipe messagePipe = messagePipeManager.getMessagePipe("test");
30+
// 随机uuid作为内容
31+
String content = UUID.randomUUID().toString();
32+
Message message = new Message(content.getBytes());
33+
// 设置元数据
34+
message.getMetadata().put("traceId", System.currentTimeMillis());
35+
messagePipe.putLast(message);
36+
}, 1000, 1000, TimeUnit.MILLISECONDS);
37+
}
38+
}

0 commit comments

Comments
 (0)