此例中我们定义中间服…1、业务介绍

消息源服务的消息不能直接推给用户侧用户与中间服务建立websocket连接中间服务再与源服务建立websocket连接源服务的消息推给中间服务中间服务再将消息推送给用户。

流程如下图

此例中我们定义中间服务A的端口为8082消息源头服务B的端口为8081方便阅读下面代码。

说明此例子只实现了中间服务的转发连接的关闭等其他逻辑并没有完善如需要请自行完善

2、中间服务实现

中间服务即为上图的中间服务A由于中间服务既要发送发给用户端消息又要接收从消息源服务B接收消息故服务A分为服务端与客户端。

websocket实现客户端使用okhttp实现会话缓存暂使用内存缓存实际项目中可置于其他缓存中

中间服务所需依赖为

dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-websocket/artifactId

/dependency

dependencygroupIdorg.projectlombok/groupIdartifactIdlombok/artifactId

/dependency

dependencygroupIdcom.squareup.okhttp3/groupIdartifactIdokhttp/artifactIdversion4.2.2/version

/dependency缓存类

ConcurrentHashMap();//存储把不同用户的客户端session信息集合

{userId,

[会话id1,会话id2,会话id3,会话id4]}public

static

massage;//为userId用于从缓存中获取对应用户的websocket

sessionprivate

response.body().string());}/***

收到消息*

收到源8081的消息取到对应userId的消息并将消息通过本地server发送给用户ObjectMapper

mapper

WSCache.connection.get(msgInfo.getUserKey());if(!CollectionUtils.isEmpty(strings)){for

(String

WSCache.clients.get(sid);session.getBasicRemote().sendText(msgInfo.getMassage());}}}

catch

RuntimeException(e);}}Overridepublic

void

ServerEndpoint(/notice/{userId})

public

UUID.randomUUID().toString();this.userId

userId;WSCache.clients.put(this.sid,session);//判断该用户是否存在会话信息不存在则添加SetString

clientSet

WSCache.connection.get(userId);if

(CollectionUtils.isEmpty(clientSet)){clientSet

new

HashSet();clientSet.add(this.sid);}else

{clientSet.add(this.sid);}WSCache.connection.put(userId,clientSet);log.info(用户{}与本地8082server建立连接,

this.userId);//todo

Request.Builder().url(ws://127.0.0.1:8081/api/notice/

new

OkHttpClient.Builder().build();WebSocket

localClientRemote

webSocketClientRemote.newWebSocket(requestRemote,

new

CommonWSClient());log.info(本地server创建本地client且本地client与远程8082server连接成功);return

userId

onClose(){WSCache.clients.remove(this.sid);System.out.println(this.sid连接断开);}/***

description:

当收到client发送的消息时执行该方法**/OnMessagepublic

void

{System.out.println(-----------收到来自用户

this.userId

error){System.out.println(error--------系统错误);error.printStackTrace();}

}websocket配置类

serverEndpointExporter(){return

new

消息源服务B只需要websocket服务用来发送消息即可其实现与中间服务A的服务端相同。

Slf4j

ServerEndpoint(/notice/{userId})

public

ConcurrentHashMap();//存储把不同用户的客户端session信息集合

{userId,

[会话id1,会话id2,会话id3,会话id4]}public

static

ConcurrentHashMap();//会话idprivate

String

userId){log.info(onOpen--session.getRequestParameterMap():{},

session.getRequestParameterMap());this.sid

UUID.randomUUID().toString();this.userId

userId;clients.put(this.sid,session);//判断该用户是否存在会话信息不存在则添加SetString

clientSet

HashSet();connection.put(userId,clientSet);}clientSet.add(this.sid);System.out.println(this.userId

用户建立连接

onClose(){clients.remove(this.sid);System.out.println(this.sid连接断开);}/***

description:

当收到客户端发送的消息时执行该方法**/OnMessagepublic

void

{System.out.println(-----------收到来自用户

this.userId

MsgInfo().setUserKey(this.userId).setMassage(服务端-

:已收到用户

message);sendMessageByUserId(this.userId,

msgInfo);}/***

error){System.out.println(error--------系统错误);error.printStackTrace();}/***

description:

connection.get(userId);//用户是否存在客户端连接if

{IteratorString

{//同步发送数据需要等上一个sendText发送完成才执行下一个发送ObjectMapper

mapper

ObjectMapper();session.getBasicRemote().sendText(mapper.writeValueAsString(msgInfo));}

catch

{e.printStackTrace();}}}}}}Scheduled(cron

0/10

testSendMessageByCron(){log.info(-----------模拟消息开始发送--------------);//模拟两个用户100和200MsgInfo

msg100

MsgInfo().setUserKey(100).setMassage(这是8081发给用户100的消息

System.currentTimeMillis());sendMessageByUserId(100,

msg100);MsgInfo

MsgInfo().setUserKey(200).setMassage(这是8081发给用户200的消息

System.currentTimeMillis());sendMessageByUserId(200,

msg200);}

打开两个该工具窗口分别模拟用户100和用户200这两个用户都连接中间服务A端口8082的服务

2、分别启动消息源服务B和中间服务A

我们模拟的消息发送已经在给用户100和用户200发送因为我们的用户100和用户200均没有与中间服务A建立连接故此时测试界面看不到消息

当我们在用户100的模拟界面点击“开启连接”后可以在右侧看到发给用户100的模拟消息

好了到这里就结束了有任何问题请积极指出此例子只是个例子并未经受任何生产的测试欢迎讨论沟通