open-soure / cim · 提交

转到一个项目

作者 crossoverJie
提交 bb53a1065ec26be2008a5553080ea72eeba8b5ae 1 个父辈 2a13ae58

:recycle: Refactoring code.清楚路由关系

内嵌 并排对比
正在显示 31 个修改的文件 包含 538 行增加 和 123 行删除
1 package com.crossoverjie.cim.client; 1 package com.crossoverjie.cim.client;
2 2
3 -import com.crossoverjie.cim.client.client.CIMClient;  
4 import com.crossoverjie.cim.client.scanner.Scan; 3 import com.crossoverjie.cim.client.scanner.Scan;
5 import org.slf4j.Logger; 4 import org.slf4j.Logger;
6 import org.slf4j.LoggerFactory; 5 import org.slf4j.LoggerFactory;
7 -import org.springframework.beans.factory.annotation.Autowired; 6 +import org.springframework.beans.factory.annotation.Value;
8 import org.springframework.boot.CommandLineRunner; 7 import org.springframework.boot.CommandLineRunner;
9 import org.springframework.boot.SpringApplication; 8 import org.springframework.boot.SpringApplication;
10 import org.springframework.boot.autoconfigure.SpringBootApplication; 9 import org.springframework.boot.autoconfigure.SpringBootApplication;
@@ -17,9 +16,8 @@ public class CIMClientApplication implements CommandLineRunner{ @@ -17,9 +16,8 @@ public class CIMClientApplication implements CommandLineRunner{
17 16
18 private final static Logger LOGGER = LoggerFactory.getLogger(CIMClientApplication.class); 17 private final static Logger LOGGER = LoggerFactory.getLogger(CIMClientApplication.class);
19 18
20 - @Autowired  
21 - private CIMClient heartbeatClient ;  
22 - 19 + @Value("${cim.user.id}")
  20 + private long userId;
23 21
24 public static void main(String[] args) { 22 public static void main(String[] args) {
25 SpringApplication.run(CIMClientApplication.class, args); 23 SpringApplication.run(CIMClientApplication.class, args);
@@ -28,7 +26,7 @@ public class CIMClientApplication implements CommandLineRunner{ @@ -28,7 +26,7 @@ public class CIMClientApplication implements CommandLineRunner{
28 26
29 @Override 27 @Override
30 public void run(String... args) throws Exception { 28 public void run(String... args) throws Exception {
31 - Scan scan = new Scan(heartbeatClient) ; 29 + Scan scan = new Scan(userId) ;
32 Thread thread = new Thread(scan); 30 Thread thread = new Thread(scan);
33 thread.setName("scan-thread"); 31 thread.setName("scan-thread");
34 thread.start(); 32 thread.start();
@@ -46,13 +46,6 @@ public class CIMClient { @@ -46,13 +46,6 @@ public class CIMClient {
46 @Value("${cim.user.userName}") 46 @Value("${cim.user.userName}")
47 private String userName; 47 private String userName;
48 48
49 -  
50 - @Value("${netty.server.port}")  
51 - private int nettyPort;  
52 -  
53 - @Value("${netty.server.host}")  
54 - private String host;  
55 -  
56 private SocketChannel channel; 49 private SocketChannel channel;
57 50
58 @Autowired 51 @Autowired
@@ -98,7 +91,7 @@ public class CIMClient { @@ -98,7 +91,7 @@ public class CIMClient {
98 .handler(new CIMClientHandleInitializer()) 91 .handler(new CIMClientHandleInitializer())
99 ; 92 ;
100 93
101 - ChannelFuture future = bootstrap.connect(cimServer.getIp(), cimServer.getPort()).sync(); 94 + ChannelFuture future = bootstrap.connect(cimServer.getIp(), cimServer.getCimServerPort()).sync();
102 if (future.isSuccess()) { 95 if (future.isSuccess()) {
103 LOGGER.info("启动 cim client 成功"); 96 LOGGER.info("启动 cim client 成功");
104 } 97 }
1 package com.crossoverjie.cim.client.controller; 1 package com.crossoverjie.cim.client.controller;
2 2
3 import com.crossoverjie.cim.client.client.CIMClient; 3 import com.crossoverjie.cim.client.client.CIMClient;
4 -import com.crossoverjie.cim.client.vo.req.SendMsgReqVO;  
5 -import com.crossoverjie.cim.client.vo.res.SendMsgResVO; 4 +import com.crossoverjie.cim.client.service.RouteRequest;
6 import com.crossoverjie.cim.client.vo.req.GoogleProtocolVO; 5 import com.crossoverjie.cim.client.vo.req.GoogleProtocolVO;
  6 +import com.crossoverjie.cim.client.vo.req.GroupReqVO;
  7 +import com.crossoverjie.cim.client.vo.req.SendMsgReqVO;
7 import com.crossoverjie.cim.client.vo.req.StringReqVO; 8 import com.crossoverjie.cim.client.vo.req.StringReqVO;
8 -import com.crossoverjie.cim.common.enums.StatusEnum;  
9 -import com.crossoverjie.cim.common.res.BaseResponse;  
10 -import com.crossoverjie.cim.client.service.RouteRequest; 9 +import com.crossoverjie.cim.client.vo.res.SendMsgResVO;
11 import com.crossoverjie.cim.common.constant.Constants; 10 import com.crossoverjie.cim.common.constant.Constants;
  11 +import com.crossoverjie.cim.common.enums.StatusEnum;
12 import com.crossoverjie.cim.common.pojo.CustomProtocol; 12 import com.crossoverjie.cim.common.pojo.CustomProtocol;
  13 +import com.crossoverjie.cim.common.res.BaseResponse;
13 import com.crossoverjie.cim.common.res.NULLBody; 14 import com.crossoverjie.cim.common.res.NULLBody;
14 import io.swagger.annotations.ApiOperation; 15 import io.swagger.annotations.ApiOperation;
15 import org.springframework.beans.factory.annotation.Autowired; 16 import org.springframework.beans.factory.annotation.Autowired;
@@ -55,7 +56,7 @@ public class IndexController { @@ -55,7 +56,7 @@ public class IndexController {
55 @ResponseBody() 56 @ResponseBody()
56 public BaseResponse<SendMsgResVO> sendMsg(@RequestBody SendMsgReqVO sendMsgReqVO){ 57 public BaseResponse<SendMsgResVO> sendMsg(@RequestBody SendMsgReqVO sendMsgReqVO){
57 BaseResponse<SendMsgResVO> res = new BaseResponse(); 58 BaseResponse<SendMsgResVO> res = new BaseResponse();
58 - heartbeatClient.sendMsg(new CustomProtocol(sendMsgReqVO.getId(),sendMsgReqVO.getMsg())) ; 59 + heartbeatClient.sendMsg(new CustomProtocol(sendMsgReqVO.getUserId(),sendMsgReqVO.getMsg())) ;
59 60
60 // 利用 actuator 来自增 61 // 利用 actuator 来自增
61 counterService.increment(Constants.COUNTER_CLIENT_PUSH_COUNT); 62 counterService.increment(Constants.COUNTER_CLIENT_PUSH_COUNT);
@@ -131,7 +132,8 @@ public class IndexController { @@ -131,7 +132,8 @@ public class IndexController {
131 public BaseResponse sendGroupMsg(@RequestBody SendMsgReqVO sendMsgReqVO) throws Exception { 132 public BaseResponse sendGroupMsg(@RequestBody SendMsgReqVO sendMsgReqVO) throws Exception {
132 BaseResponse<NULLBody> res = new BaseResponse(); 133 BaseResponse<NULLBody> res = new BaseResponse();
133 134
134 - routeRequest.sendGroupMsg(sendMsgReqVO.getMsg()) ; 135 + GroupReqVO groupReqVO = new GroupReqVO(sendMsgReqVO.getUserId(),sendMsgReqVO.getMsg()) ;
  136 + routeRequest.sendGroupMsg(groupReqVO) ;
135 137
136 counterService.increment(Constants.COUNTER_SERVER_PUSH_COUNT); 138 counterService.increment(Constants.COUNTER_SERVER_PUSH_COUNT);
137 139
1 package com.crossoverjie.cim.client.scanner; 1 package com.crossoverjie.cim.client.scanner;
2 2
3 import com.crossoverjie.cim.client.client.CIMClient; 3 import com.crossoverjie.cim.client.client.CIMClient;
  4 +import com.crossoverjie.cim.client.service.RouteRequest;
  5 +import com.crossoverjie.cim.client.util.SpringBeanFactory;
4 import com.crossoverjie.cim.client.vo.req.GoogleProtocolVO; 6 import com.crossoverjie.cim.client.vo.req.GoogleProtocolVO;
  7 +import com.crossoverjie.cim.client.vo.req.GroupReqVO;
5 import org.slf4j.Logger; 8 import org.slf4j.Logger;
6 import org.slf4j.LoggerFactory; 9 import org.slf4j.LoggerFactory;
7 10
@@ -14,31 +17,49 @@ import java.util.Scanner; @@ -14,31 +17,49 @@ import java.util.Scanner;
14 * Date: 2018/12/21 16:44 17 * Date: 2018/12/21 16:44
15 * @since JDK 1.8 18 * @since JDK 1.8
16 */ 19 */
17 -public class Scan implements Runnable{ 20 +public class Scan implements Runnable {
18 21
19 private final static Logger LOGGER = LoggerFactory.getLogger(Scan.class); 22 private final static Logger LOGGER = LoggerFactory.getLogger(Scan.class);
20 23
21 - private CIMClient heartbeatClient ; 24 + private CIMClient heartbeatClient;
22 25
23 - public Scan(CIMClient heartbeatClient) {  
24 - this.heartbeatClient = heartbeatClient; 26 + private RouteRequest routeRequest;
  27 +
  28 + private Long userId ;
  29 +
  30 + public Scan(Long userId) {
  31 + this.userId = userId ;
  32 + this.heartbeatClient = SpringBeanFactory.getBean(CIMClient.class);
  33 + this.routeRequest = SpringBeanFactory.getBean(RouteRequest.class);
25 } 34 }
26 35
27 @Override 36 @Override
28 public void run() { 37 public void run() {
29 Scanner sc = new Scanner(System.in); 38 Scanner sc = new Scanner(System.in);
30 - String[] totalMsg ;  
31 - GoogleProtocolVO vo ;  
32 - while (true){  
33 - String msg = sc.nextLine() ; 39 + String[] totalMsg;
  40 + GoogleProtocolVO vo;
  41 + while (true) {
  42 + String msg = sc.nextLine();
34 43
  44 + //单聊
35 totalMsg = msg.split(" "); 45 totalMsg = msg.split(" ");
36 -  
37 - vo = new GoogleProtocolVO() ; 46 + if (totalMsg.length > 1) {
  47 + vo = new GoogleProtocolVO();
38 vo.setRequestId(Integer.parseInt(totalMsg[0])); 48 vo.setRequestId(Integer.parseInt(totalMsg[0]));
39 vo.setMsg(totalMsg[1]); 49 vo.setMsg(totalMsg[1]);
40 - heartbeatClient.sendGoogleProtocolMsg(vo) ;  
41 - LOGGER.info("scan =[{}]",msg); 50 + heartbeatClient.sendGoogleProtocolMsg(vo);
  51 + } else {
  52 + //群聊
  53 + try {
  54 + GroupReqVO groupReqVO = new GroupReqVO(userId,msg) ;
  55 + routeRequest.sendGroupMsg(groupReqVO) ;
  56 + } catch (Exception e) {
  57 + LOGGER.error("Exception", e);
  58 + }
  59 + }
  60 +
  61 +
  62 + LOGGER.info("scan =[{}]", msg);
42 } 63 }
43 } 64 }
44 } 65 }
1 package com.crossoverjie.cim.client.service; 1 package com.crossoverjie.cim.client.service;
2 2
  3 +import com.crossoverjie.cim.client.vo.req.GroupReqVO;
3 import com.crossoverjie.cim.client.vo.req.LoginReqVO; 4 import com.crossoverjie.cim.client.vo.req.LoginReqVO;
4 import com.crossoverjie.cim.client.vo.res.CIMServerResVO; 5 import com.crossoverjie.cim.client.vo.res.CIMServerResVO;
5 6
@@ -14,10 +15,10 @@ public interface RouteRequest { @@ -14,10 +15,10 @@ public interface RouteRequest {
14 15
15 /** 16 /**
16 * 群发消息 17 * 群发消息
17 - * @param msg 消息 18 + * @param groupReqVO 消息
18 * @throws Exception 19 * @throws Exception
19 */ 20 */
20 - void sendGroupMsg(String msg) throws Exception; 21 + void sendGroupMsg(GroupReqVO groupReqVO) throws Exception;
21 22
22 /** 23 /**
23 * 获取服务器 24 * 获取服务器
@@ -3,6 +3,7 @@ package com.crossoverjie.cim.client.service.impl; @@ -3,6 +3,7 @@ package com.crossoverjie.cim.client.service.impl;
3 import com.alibaba.fastjson.JSON; 3 import com.alibaba.fastjson.JSON;
4 import com.alibaba.fastjson.JSONObject; 4 import com.alibaba.fastjson.JSONObject;
5 import com.crossoverjie.cim.client.service.RouteRequest; 5 import com.crossoverjie.cim.client.service.RouteRequest;
  6 +import com.crossoverjie.cim.client.vo.req.GroupReqVO;
6 import com.crossoverjie.cim.client.vo.req.LoginReqVO; 7 import com.crossoverjie.cim.client.vo.req.LoginReqVO;
7 import com.crossoverjie.cim.client.vo.res.CIMServerResVO; 8 import com.crossoverjie.cim.client.vo.res.CIMServerResVO;
8 import com.crossoverjie.cim.common.enums.StatusEnum; 9 import com.crossoverjie.cim.common.enums.StatusEnum;
@@ -39,10 +40,11 @@ public class RouteRequestImpl implements RouteRequest { @@ -39,10 +40,11 @@ public class RouteRequestImpl implements RouteRequest {
39 private String serverRouteRequestUrl; 40 private String serverRouteRequestUrl;
40 41
41 @Override 42 @Override
42 - public void sendGroupMsg(String msg) throws Exception { 43 + public void sendGroupMsg(GroupReqVO groupReqVO) throws Exception {
43 44
44 JSONObject jsonObject = new JSONObject(); 45 JSONObject jsonObject = new JSONObject();
45 - jsonObject.put("msg",msg); 46 + jsonObject.put("msg",groupReqVO.getMsg());
  47 + jsonObject.put("userId",groupReqVO.getUserId());
46 RequestBody requestBody = RequestBody.create(mediaType,jsonObject.toString()); 48 RequestBody requestBody = RequestBody.create(mediaType,jsonObject.toString());
47 49
48 Request request = new Request.Builder() 50 Request request = new Request.Builder()
1 -package com.crossoverjie.cim.route.vo.req; 1 +package com.crossoverjie.cim.client.vo.req;
2 2
3 import com.crossoverjie.cim.common.req.BaseRequest; 3 import com.crossoverjie.cim.common.req.BaseRequest;
4 import io.swagger.annotations.ApiModelProperty; 4 import io.swagger.annotations.ApiModelProperty;
@@ -14,10 +14,20 @@ import javax.validation.constraints.NotNull; @@ -14,10 +14,20 @@ import javax.validation.constraints.NotNull;
14 */ 14 */
15 public class GroupReqVO extends BaseRequest { 15 public class GroupReqVO extends BaseRequest {
16 16
  17 + @NotNull(message = "userId 不能为空")
  18 + @ApiModelProperty(required = true, value = "消息发送者的 userId", example = "1545574049323")
  19 + private Long userId ;
  20 +
  21 +
17 @NotNull(message = "msg 不能为空") 22 @NotNull(message = "msg 不能为空")
18 @ApiModelProperty(required = true, value = "msg", example = "hello") 23 @ApiModelProperty(required = true, value = "msg", example = "hello")
19 private String msg ; 24 private String msg ;
20 25
  26 + public GroupReqVO(Long userId, String msg) {
  27 + this.userId = userId;
  28 + this.msg = msg;
  29 + }
  30 +
21 public String getMsg() { 31 public String getMsg() {
22 return msg; 32 return msg;
23 } 33 }
@@ -26,10 +36,19 @@ public class GroupReqVO extends BaseRequest { @@ -26,10 +36,19 @@ public class GroupReqVO extends BaseRequest {
26 this.msg = msg; 36 this.msg = msg;
27 } 37 }
28 38
  39 + public Long getUserId() {
  40 + return userId;
  41 + }
  42 +
  43 + public void setUserId(Long userId) {
  44 + this.userId = userId;
  45 + }
  46 +
29 @Override 47 @Override
30 public String toString() { 48 public String toString() {
31 - return "GroupRequest{" +  
32 - "msg='" + msg + '\'' + 49 + return "GroupReqVO{" +
  50 + "userId=" + userId +
  51 + ", msg='" + msg + '\'' +
33 "} " + super.toString(); 52 "} " + super.toString();
34 } 53 }
35 } 54 }
@@ -18,9 +18,9 @@ public class SendMsgReqVO extends BaseRequest { @@ -18,9 +18,9 @@ public class SendMsgReqVO extends BaseRequest {
18 @ApiModelProperty(required = true, value = "msg", example = "hello") 18 @ApiModelProperty(required = true, value = "msg", example = "hello")
19 private String msg ; 19 private String msg ;
20 20
21 - @NotNull(message = "id 不能为空")  
22 - @ApiModelProperty(required = true, value = "id", example = "11")  
23 - private long id ; 21 + @NotNull(message = "userId 不能为空")
  22 + @ApiModelProperty(required = true, value = "userId", example = "11")
  23 + private Long userId ;
24 24
25 public String getMsg() { 25 public String getMsg() {
26 return msg; 26 return msg;
@@ -30,11 +30,11 @@ public class SendMsgReqVO extends BaseRequest { @@ -30,11 +30,11 @@ public class SendMsgReqVO extends BaseRequest {
30 this.msg = msg; 30 this.msg = msg;
31 } 31 }
32 32
33 - public long getId() {  
34 - return id; 33 + public long getUserId() {
  34 + return userId;
35 } 35 }
36 36
37 - public void setId(long id) {  
38 - this.id = id; 37 + public void setUserId(long userId) {
  38 + this.userId = userId;
39 } 39 }
40 } 40 }
@@ -60,9 +60,9 @@ public class CIMServerResVO implements Serializable { @@ -60,9 +60,9 @@ public class CIMServerResVO implements Serializable {
60 * ip : 127.0.0.1 60 * ip : 127.0.0.1
61 * port : 8081 61 * port : 8081
62 */ 62 */
63 -  
64 - private String ip;  
65 - private int port; 63 + private String ip ;
  64 + private Integer cimServerPort;
  65 + private Integer httpPort;
66 66
67 public String getIp() { 67 public String getIp() {
68 return ip; 68 return ip;
@@ -72,19 +72,28 @@ public class CIMServerResVO implements Serializable { @@ -72,19 +72,28 @@ public class CIMServerResVO implements Serializable {
72 this.ip = ip; 72 this.ip = ip;
73 } 73 }
74 74
75 - public int getPort() {  
76 - return port; 75 + public Integer getCimServerPort() {
  76 + return cimServerPort;
  77 + }
  78 +
  79 + public void setCimServerPort(Integer cimServerPort) {
  80 + this.cimServerPort = cimServerPort;
  81 + }
  82 +
  83 + public Integer getHttpPort() {
  84 + return httpPort;
77 } 85 }
78 86
79 - public void setPort(int port) {  
80 - this.port = port; 87 + public void setHttpPort(Integer httpPort) {
  88 + this.httpPort = httpPort;
81 } 89 }
82 90
83 @Override 91 @Override
84 public String toString() { 92 public String toString() {
85 - return "DataBodyBean{" + 93 + return "ServerInfo{" +
86 "ip='" + ip + '\'' + 94 "ip='" + ip + '\'' +
87 - ", port=" + port + 95 + ", cimServerPort=" + cimServerPort +
  96 + ", httpPort=" + httpPort +
88 '}'; 97 '}';
89 } 98 }
90 } 99 }
@@ -6,9 +6,6 @@ server.port=8082 @@ -6,9 +6,6 @@ server.port=8082
6 # 是否打开swagger 6 # 是否打开swagger
7 swagger.enable = true 7 swagger.enable = true
8 8
9 -netty.server.host=127.0.0.1  
10 -netty.server.port=11211  
11 -  
12 logging.level.root=info 9 logging.level.root=info
13 10
14 # 群发消息 11 # 群发消息
@@ -23,5 +23,8 @@ public class CommonTest { @@ -23,5 +23,8 @@ public class CommonTest {
23 23
24 System.out.println(cimServerResVO.toString()); 24 System.out.println(cimServerResVO.toString());
25 25
  26 + String text = "nihaoaaa" ;
  27 + String[] split = text.split(" ");
  28 + System.out.println(split.length);
26 } 29 }
27 } 30 }
@@ -3,6 +3,7 @@ package com.crossoverjie.cim.route.config; @@ -3,6 +3,7 @@ package com.crossoverjie.cim.route.config;
3 import com.google.common.cache.CacheBuilder; 3 import com.google.common.cache.CacheBuilder;
4 import com.google.common.cache.CacheLoader; 4 import com.google.common.cache.CacheLoader;
5 import com.google.common.cache.LoadingCache; 5 import com.google.common.cache.LoadingCache;
  6 +import okhttp3.OkHttpClient;
6 import org.I0Itec.zkclient.ZkClient; 7 import org.I0Itec.zkclient.ZkClient;
7 import org.springframework.beans.factory.annotation.Autowired; 8 import org.springframework.beans.factory.annotation.Autowired;
8 import org.springframework.context.annotation.Bean; 9 import org.springframework.context.annotation.Bean;
@@ -12,6 +13,8 @@ import org.springframework.data.redis.core.RedisTemplate; @@ -12,6 +13,8 @@ import org.springframework.data.redis.core.RedisTemplate;
12 import org.springframework.data.redis.core.StringRedisTemplate; 13 import org.springframework.data.redis.core.StringRedisTemplate;
13 import org.springframework.data.redis.serializer.StringRedisSerializer; 14 import org.springframework.data.redis.serializer.StringRedisSerializer;
14 15
  16 +import java.util.concurrent.TimeUnit;
  17 +
15 /** 18 /**
16 * Function: 19 * Function:
17 * 20 *
@@ -55,4 +58,19 @@ public class BeanConfig { @@ -55,4 +58,19 @@ public class BeanConfig {
55 redisTemplate.afterPropertiesSet(); 58 redisTemplate.afterPropertiesSet();
56 return redisTemplate; 59 return redisTemplate;
57 } 60 }
  61 +
  62 +
  63 + /**
  64 + * http client
  65 + * @return okHttp
  66 + */
  67 + @Bean
  68 + public OkHttpClient okHttpClient() {
  69 + OkHttpClient.Builder builder = new OkHttpClient.Builder();
  70 + builder.connectTimeout(30, TimeUnit.SECONDS)
  71 + .readTimeout(10, TimeUnit.SECONDS)
  72 + .writeTimeout(10,TimeUnit.SECONDS)
  73 + .retryOnConnectionFailure(true);
  74 + return builder.build();
  75 + }
58 } 76 }
@@ -5,7 +5,7 @@ import com.crossoverjie.cim.common.res.BaseResponse; @@ -5,7 +5,7 @@ import com.crossoverjie.cim.common.res.BaseResponse;
5 import com.crossoverjie.cim.common.res.NULLBody; 5 import com.crossoverjie.cim.common.res.NULLBody;
6 import com.crossoverjie.cim.route.cache.ServerCache; 6 import com.crossoverjie.cim.route.cache.ServerCache;
7 import com.crossoverjie.cim.route.service.AccountService; 7 import com.crossoverjie.cim.route.service.AccountService;
8 -import com.crossoverjie.cim.route.vo.req.GroupReqVO; 8 +import com.crossoverjie.cim.route.vo.req.ChatReqVO;
9 import com.crossoverjie.cim.route.vo.req.LoginReqVO; 9 import com.crossoverjie.cim.route.vo.req.LoginReqVO;
10 import com.crossoverjie.cim.route.vo.req.P2PReqVO; 10 import com.crossoverjie.cim.route.vo.req.P2PReqVO;
11 import com.crossoverjie.cim.route.vo.req.RegisterInfoReqVO; 11 import com.crossoverjie.cim.route.vo.req.RegisterInfoReqVO;
@@ -21,6 +21,8 @@ import org.springframework.web.bind.annotation.RequestMapping; @@ -21,6 +21,8 @@ import org.springframework.web.bind.annotation.RequestMapping;
21 import org.springframework.web.bind.annotation.RequestMethod; 21 import org.springframework.web.bind.annotation.RequestMethod;
22 import org.springframework.web.bind.annotation.ResponseBody; 22 import org.springframework.web.bind.annotation.ResponseBody;
23 23
  24 +import java.util.Map;
  25 +
24 /** 26 /**
25 * Function: 27 * Function:
26 * 28 *
@@ -42,11 +44,43 @@ public class RouteController { @@ -42,11 +44,43 @@ public class RouteController {
42 @ApiOperation("群聊 API") 44 @ApiOperation("群聊 API")
43 @RequestMapping(value = "groupRoute", method = RequestMethod.POST) 45 @RequestMapping(value = "groupRoute", method = RequestMethod.POST)
44 @ResponseBody() 46 @ResponseBody()
45 - public BaseResponse<NULLBody> groupRoute(@RequestBody GroupReqVO groupReqVO) { 47 + public BaseResponse<NULLBody> groupRoute(@RequestBody ChatReqVO groupReqVO) throws Exception {
46 BaseResponse<NULLBody> res = new BaseResponse(); 48 BaseResponse<NULLBody> res = new BaseResponse();
47 49
48 LOGGER.info("msg=[{}]", groupReqVO.toString()); 50 LOGGER.info("msg=[{}]", groupReqVO.toString());
49 51
  52 + //获取所有的推送列表
  53 + Map<Long, CIMServerResVO> serverResVOMap = accountService.loadRouteRelated();
  54 + for (Map.Entry<Long, CIMServerResVO> cimServerResVOEntry : serverResVOMap.entrySet()) {
  55 + Long userId = cimServerResVOEntry.getKey();
  56 + CIMServerResVO value = cimServerResVOEntry.getValue();
  57 + if (userId.equals(groupReqVO.getUserId())){
  58 + //过滤掉自己
  59 + LOGGER.info("过滤掉了发送者 userId={}",groupReqVO.getUserId());
  60 + continue;
  61 + }
  62 +
  63 + //推送消息
  64 + String url = "http://" + value.getIp() + ":" + value.getHttpPort() + "/sendMsg" ;
  65 + ChatReqVO vo = new ChatReqVO(userId,groupReqVO.getMsg()) ;
  66 + accountService.pushMsg(url,vo);
  67 +
  68 + }
  69 +
  70 + res.setCode(StatusEnum.SUCCESS.getCode());
  71 + res.setMessage(StatusEnum.SUCCESS.getMessage());
  72 + return res;
  73 + }
  74 +
  75 + @ApiOperation("客户端下线")
  76 + @RequestMapping(value = "offLine", method = RequestMethod.POST)
  77 + @ResponseBody()
  78 + public BaseResponse<NULLBody> offLine(@RequestBody ChatReqVO groupReqVO) throws Exception {
  79 + BaseResponse<NULLBody> res = new BaseResponse();
  80 +
  81 + LOGGER.info("下线用户[{}]", groupReqVO.toString());
  82 + accountService.offLine(groupReqVO.getUserId());
  83 +
50 res.setCode(StatusEnum.SUCCESS.getCode()); 84 res.setCode(StatusEnum.SUCCESS.getCode());
51 res.setMessage(StatusEnum.SUCCESS.getMessage()); 85 res.setMessage(StatusEnum.SUCCESS.getMessage());
52 return res; 86 return res;
@@ -86,7 +120,7 @@ public class RouteController { @@ -86,7 +120,7 @@ public class RouteController {
86 if (login) { 120 if (login) {
87 String server = serverCache.selectServer(); 121 String server = serverCache.selectServer();
88 String[] serverInfo = server.split(":"); 122 String[] serverInfo = server.split(":");
89 - CIMServerResVO vo = new CIMServerResVO(serverInfo[0], Integer.parseInt(serverInfo[1])); 123 + CIMServerResVO vo = new CIMServerResVO(serverInfo[0], Integer.parseInt(serverInfo[1]),Integer.parseInt(serverInfo[2]));
90 124
91 //保存路由信息 125 //保存路由信息
92 accountService.saveRouteInfo(loginReqVO,server); 126 accountService.saveRouteInfo(loginReqVO,server);
1 package com.crossoverjie.cim.route.service; 1 package com.crossoverjie.cim.route.service;
2 2
  3 +import com.crossoverjie.cim.route.vo.req.ChatReqVO;
3 import com.crossoverjie.cim.route.vo.req.LoginReqVO; 4 import com.crossoverjie.cim.route.vo.req.LoginReqVO;
  5 +import com.crossoverjie.cim.route.vo.res.CIMServerResVO;
4 import com.crossoverjie.cim.route.vo.res.RegisterInfoResVO; 6 import com.crossoverjie.cim.route.vo.res.RegisterInfoResVO;
5 7
  8 +import java.util.Map;
  9 +
6 /** 10 /**
7 * Function: 账户服务 11 * Function: 账户服务
8 * 12 *
@@ -35,4 +39,26 @@ public interface AccountService { @@ -35,4 +39,26 @@ public interface AccountService {
35 * @throws Exception 39 * @throws Exception
36 */ 40 */
37 void saveRouteInfo(LoginReqVO loginReqVO ,String msg) throws Exception ; 41 void saveRouteInfo(LoginReqVO loginReqVO ,String msg) throws Exception ;
  42 +
  43 + /**
  44 + * 加载所有的路有关系
  45 + * @return 所有的路由关系
  46 + */
  47 + Map<Long,CIMServerResVO> loadRouteRelated() ;
  48 +
  49 +
  50 + /**
  51 + * 推送消息
  52 + * @param url url
  53 + * @param groupReqVO 消息
  54 + * @throws Exception
  55 + */
  56 + void pushMsg(String url,ChatReqVO groupReqVO) throws Exception;
  57 +
  58 + /**
  59 + * 用户下线
  60 + * @param userId 下线用户ID
  61 + * @throws Exception
  62 + */
  63 + void offLine(Long userId) throws Exception;
38 } 64 }
1 package com.crossoverjie.cim.route.service.impl; 1 package com.crossoverjie.cim.route.service.impl;
2 2
  3 +import com.alibaba.fastjson.JSONObject;
3 import com.crossoverjie.cim.route.service.AccountService; 4 import com.crossoverjie.cim.route.service.AccountService;
  5 +import com.crossoverjie.cim.route.vo.req.ChatReqVO;
4 import com.crossoverjie.cim.route.vo.req.LoginReqVO; 6 import com.crossoverjie.cim.route.vo.req.LoginReqVO;
  7 +import com.crossoverjie.cim.route.vo.res.CIMServerResVO;
5 import com.crossoverjie.cim.route.vo.res.RegisterInfoResVO; 8 import com.crossoverjie.cim.route.vo.res.RegisterInfoResVO;
  9 +import okhttp3.*;
6 import org.slf4j.Logger; 10 import org.slf4j.Logger;
7 import org.slf4j.LoggerFactory; 11 import org.slf4j.LoggerFactory;
8 import org.springframework.beans.factory.annotation.Autowired; 12 import org.springframework.beans.factory.annotation.Autowired;
  13 +import org.springframework.data.redis.connection.RedisConnection;
  14 +import org.springframework.data.redis.core.Cursor;
9 import org.springframework.data.redis.core.RedisTemplate; 15 import org.springframework.data.redis.core.RedisTemplate;
  16 +import org.springframework.data.redis.core.ScanOptions;
10 import org.springframework.stereotype.Service; 17 import org.springframework.stereotype.Service;
11 18
  19 +import java.io.IOException;
  20 +import java.nio.charset.StandardCharsets;
  21 +import java.util.HashMap;
  22 +import java.util.Map;
  23 +
12 import static com.crossoverjie.cim.route.constant.Constant.ACCOUNT_PREFIX; 24 import static com.crossoverjie.cim.route.constant.Constant.ACCOUNT_PREFIX;
13 import static com.crossoverjie.cim.route.constant.Constant.ROUTE_PREFIX; 25 import static com.crossoverjie.cim.route.constant.Constant.ROUTE_PREFIX;
14 26
@@ -26,41 +38,103 @@ public class AccountServiceRedisImpl implements AccountService { @@ -26,41 +38,103 @@ public class AccountServiceRedisImpl implements AccountService {
26 @Autowired 38 @Autowired
27 private RedisTemplate<String, String> redisTemplate; 39 private RedisTemplate<String, String> redisTemplate;
28 40
  41 + @Autowired
  42 + private OkHttpClient okHttpClient;
  43 +
  44 + private MediaType mediaType = MediaType.parse("application/json");
  45 +
29 @Override 46 @Override
30 public RegisterInfoResVO register(RegisterInfoResVO info) { 47 public RegisterInfoResVO register(RegisterInfoResVO info) {
31 String key = ACCOUNT_PREFIX + info.getUserId(); 48 String key = ACCOUNT_PREFIX + info.getUserId();
32 49
33 - String name = redisTemplate.opsForValue().get(info.getUserName()) ;  
34 - if (null == name){ 50 + String name = redisTemplate.opsForValue().get(info.getUserName());
  51 + if (null == name) {
35 //为了方便查询，冗余一份 52 //为了方便查询，冗余一份
36 redisTemplate.opsForValue().set(key, info.getUserName()); 53 redisTemplate.opsForValue().set(key, info.getUserName());
37 - redisTemplate.opsForValue().set(info.getUserName(),key);  
38 - }else { 54 + redisTemplate.opsForValue().set(info.getUserName(), key);
  55 + } else {
39 long userId = Long.parseLong(name.split(":")[1]); 56 long userId = Long.parseLong(name.split(":")[1]);
40 info.setUserId(userId); 57 info.setUserId(userId);
41 info.setUserName(info.getUserName()); 58 info.setUserName(info.getUserName());
42 } 59 }
43 60
44 - return info ; 61 + return info;
45 } 62 }
46 63
47 @Override 64 @Override
48 public boolean login(LoginReqVO loginReqVO) throws Exception { 65 public boolean login(LoginReqVO loginReqVO) throws Exception {
49 String key = ACCOUNT_PREFIX + loginReqVO.getUserId(); 66 String key = ACCOUNT_PREFIX + loginReqVO.getUserId();
50 String userName = redisTemplate.opsForValue().get(key); 67 String userName = redisTemplate.opsForValue().get(key);
51 - if (null == userName){  
52 - return false ; 68 + if (null == userName) {
  69 + return false;
  70 + }
  71 +
  72 + if (!userName.equals(loginReqVO.getUserName())) {
  73 + return false;
  74 + }
  75 + return true;
  76 + }
  77 +
  78 + @Override
  79 + public void saveRouteInfo(LoginReqVO loginReqVO, String msg) throws Exception {
  80 + String key = ROUTE_PREFIX + loginReqVO.getUserId();
  81 + redisTemplate.opsForValue().set(key, msg);
  82 + }
  83 +
  84 + @Override
  85 + public Map<Long, CIMServerResVO> loadRouteRelated() {
  86 +
  87 + Map<Long, CIMServerResVO> routes = new HashMap<>(64);
  88 +
  89 + RedisConnection connection = redisTemplate.getConnectionFactory().getConnection();
  90 + ScanOptions options = ScanOptions.scanOptions()
  91 + .match(ROUTE_PREFIX + "*")
  92 + .build();
  93 + Cursor<byte[]> scan = connection.scan(options);
  94 +
  95 + while (scan.hasNext()) {
  96 + byte[] next = scan.next();
  97 + String key = new String(next, StandardCharsets.UTF_8);
  98 + LOGGER.info("key={}", key);
  99 + parseServerInfo(routes, key);
  100 +
53 } 101 }
54 102
55 - if (!userName.equals(loginReqVO.getUserName())){  
56 - return false ; 103 + return routes;
  104 + }
  105 +
  106 + private void parseServerInfo(Map<Long, CIMServerResVO> routes, String key) {
  107 + long userId = Long.valueOf(key.split(":")[1]);
  108 + String value = redisTemplate.opsForValue().get(key);
  109 + String[] server = value.split(":");
  110 + CIMServerResVO cimServerResVO = new CIMServerResVO(server[0], Integer.parseInt(server[1]), Integer.parseInt(server[2]));
  111 + routes.put(userId, cimServerResVO);
  112 + }
  113 +
  114 +
  115 + @Override
  116 + public void pushMsg(String url, ChatReqVO groupReqVO) throws Exception {
  117 + Long userId = groupReqVO.getUserId();
  118 + String userName = redisTemplate.opsForValue().get(ACCOUNT_PREFIX + userId);
  119 +
  120 + JSONObject jsonObject = new JSONObject();
  121 + jsonObject.put("msg", userName + ":【" + groupReqVO.getMsg() + "】");
  122 + jsonObject.put("userId", userId);
  123 + RequestBody requestBody = RequestBody.create(mediaType, jsonObject.toString());
  124 +
  125 + Request request = new Request.Builder()
  126 + .url(url)
  127 + .post(requestBody)
  128 + .build();
  129 +
  130 + Response response = okHttpClient.newCall(request).execute();
  131 + if (!response.isSuccessful()) {
  132 + throw new IOException("Unexpected code " + response);
57 } 133 }
58 - return true ;  
59 } 134 }
60 135
61 @Override 136 @Override
62 - public void saveRouteInfo(LoginReqVO loginReqVO,String msg) throws Exception {  
63 - String key = ROUTE_PREFIX + loginReqVO.getUserId() ;  
64 - redisTemplate.opsForValue().set(key,msg) ; 137 + public void offLine(Long userId) throws Exception {
  138 + redisTemplate.delete(ROUTE_PREFIX + userId) ;
65 } 139 }
66 } 140 }
  1 +package com.crossoverjie.cim.route.vo.req;
  2 +
  3 +import com.crossoverjie.cim.common.req.BaseRequest;
  4 +import io.swagger.annotations.ApiModelProperty;
  5 +
  6 +import javax.validation.constraints.NotNull;
  7 +
  8 +/**
  9 + * Function: Google Protocol 编解码发送
  10 + *
  11 + * @author crossoverJie
  12 + * Date: 2018/05/21 15:56
  13 + * @since JDK 1.8
  14 + */
  15 +public class ChatReqVO extends BaseRequest {
  16 +
  17 + @NotNull(message = "userId 不能为空")
  18 + @ApiModelProperty(required = true, value = "userId", example = "1545574049323")
  19 + private Long userId ;
  20 +
  21 +
  22 + @NotNull(message = "msg 不能为空")
  23 + @ApiModelProperty(required = true, value = "msg", example = "hello")
  24 + private String msg ;
  25 +
  26 + public ChatReqVO() {
  27 + }
  28 +
  29 + public ChatReqVO(Long userId, String msg) {
  30 + this.userId = userId;
  31 + this.msg = msg;
  32 + }
  33 +
  34 + public String getMsg() {
  35 + return msg;
  36 + }
  37 +
  38 + public void setMsg(String msg) {
  39 + this.msg = msg;
  40 + }
  41 +
  42 + public Long getUserId() {
  43 + return userId;
  44 + }
  45 +
  46 + public void setUserId(Long userId) {
  47 + this.userId = userId;
  48 + }
  49 +
  50 + @Override
  51 + public String toString() {
  52 + return "GroupReqVO{" +
  53 + "userId=" + userId +
  54 + ", msg='" + msg + '\'' +
  55 + "} " + super.toString();
  56 + }
  57 +}
@@ -12,11 +12,13 @@ import java.io.Serializable; @@ -12,11 +12,13 @@ import java.io.Serializable;
12 public class CIMServerResVO implements Serializable { 12 public class CIMServerResVO implements Serializable {
13 13
14 private String ip ; 14 private String ip ;
15 - private Integer port; 15 + private Integer cimServerPort;
  16 + private Integer httpPort;
16 17
17 - public CIMServerResVO(String ip, Integer port) { 18 + public CIMServerResVO(String ip, Integer cimServerPort, Integer httpPort) {
18 this.ip = ip; 19 this.ip = ip;
19 - this.port = port; 20 + this.cimServerPort = cimServerPort;
  21 + this.httpPort = httpPort;
20 } 22 }
21 23
22 public String getIp() { 24 public String getIp() {
@@ -27,19 +29,19 @@ public class CIMServerResVO implements Serializable { @@ -27,19 +29,19 @@ public class CIMServerResVO implements Serializable {
27 this.ip = ip; 29 this.ip = ip;
28 } 30 }
29 31
30 - public Integer getPort() {  
31 - return port; 32 + public Integer getCimServerPort() {
  33 + return cimServerPort;
32 } 34 }
33 35
34 - public void setPort(Integer port) {  
35 - this.port = port; 36 + public void setCimServerPort(Integer cimServerPort) {
  37 + this.cimServerPort = cimServerPort;
36 } 38 }
37 39
38 - @Override  
39 - public String toString() {  
40 - return "CIMServerResVO{" +  
41 - "ip='" + ip + '\'' +  
42 - ", port=" + port +  
43 - '}'; 40 + public Integer getHttpPort() {
  41 + return httpPort;
  42 + }
  43 +
  44 + public void setHttpPort(Integer httpPort) {
  45 + this.httpPort = httpPort;
44 } 46 }
45 } 47 }
  1 +package com.crossoverjie.cim.route.service.impl;
  2 +
  3 +import com.alibaba.fastjson.JSON;
  4 +import com.crossoverjie.cim.route.RouteApplication;
  5 +import com.crossoverjie.cim.route.service.AccountService;
  6 +import com.crossoverjie.cim.route.vo.res.CIMServerResVO;
  7 +import org.junit.Test;
  8 +import org.junit.runner.RunWith;
  9 +import org.slf4j.Logger;
  10 +import org.slf4j.LoggerFactory;
  11 +import org.springframework.beans.factory.annotation.Autowired;
  12 +import org.springframework.boot.test.context.SpringBootTest;
  13 +import org.springframework.test.context.junit4.SpringRunner;
  14 +
  15 +import java.util.Map;
  16 +
  17 +@SpringBootTest(classes = RouteApplication.class)
  18 +@RunWith(SpringRunner.class)
  19 +public class AccountServiceRedisImplTest {
  20 +
  21 + private final static Logger LOGGER = LoggerFactory.getLogger(AccountServiceRedisImplTest.class);
  22 +
  23 + @Autowired
  24 + private AccountService accountService ;
  25 +
  26 + @Test
  27 + public void loadRouteRelated() throws Exception {
  28 + Map<Long, CIMServerResVO> longCIMServerResVOMap = accountService.loadRouteRelated();
  29 + LOGGER.info("longCIMServerResVOMap={}" , JSON.toJSONString(longCIMServerResVOMap));
  30 + }
  31 +
  32 +}
@@ -5,6 +5,7 @@ import com.crossoverjie.cim.server.kit.RegistryZK; @@ -5,6 +5,7 @@ import com.crossoverjie.cim.server.kit.RegistryZK;
5 import org.slf4j.Logger; 5 import org.slf4j.Logger;
6 import org.slf4j.LoggerFactory; 6 import org.slf4j.LoggerFactory;
7 import org.springframework.beans.factory.annotation.Autowired; 7 import org.springframework.beans.factory.annotation.Autowired;
  8 +import org.springframework.beans.factory.annotation.Value;
8 import org.springframework.boot.CommandLineRunner; 9 import org.springframework.boot.CommandLineRunner;
9 import org.springframework.boot.SpringApplication; 10 import org.springframework.boot.SpringApplication;
10 import org.springframework.boot.autoconfigure.SpringBootApplication; 11 import org.springframework.boot.autoconfigure.SpringBootApplication;
@@ -22,6 +23,9 @@ public class CIMServerApplication implements CommandLineRunner{ @@ -22,6 +23,9 @@ public class CIMServerApplication implements CommandLineRunner{
22 @Autowired 23 @Autowired
23 private AppConfiguration appConfiguration ; 24 private AppConfiguration appConfiguration ;
24 25
  26 + @Value("${server.port}")
  27 + private int httpPort ;
  28 +
25 public static void main(String[] args) { 29 public static void main(String[] args) {
26 SpringApplication.run(CIMServerApplication.class, args); 30 SpringApplication.run(CIMServerApplication.class, args);
27 LOGGER.info("启动 Server 成功"); 31 LOGGER.info("启动 Server 成功");
@@ -31,7 +35,7 @@ public class CIMServerApplication implements CommandLineRunner{ @@ -31,7 +35,7 @@ public class CIMServerApplication implements CommandLineRunner{
31 public void run(String... args) throws Exception { 35 public void run(String... args) throws Exception {
32 //获得本机IP 36 //获得本机IP
33 String addr = InetAddress.getLocalHost().getHostAddress(); 37 String addr = InetAddress.getLocalHost().getHostAddress();
34 - Thread thread = new Thread(new RegistryZK(addr, appConfiguration.getPort())); 38 + Thread thread = new Thread(new RegistryZK(addr, appConfiguration.getCimServerPort(),httpPort));
35 thread.setName("registry-zk"); 39 thread.setName("registry-zk");
36 thread.start() ; 40 thread.start() ;
37 } 41 }
@@ -22,15 +22,18 @@ public class AppConfiguration { @@ -22,15 +22,18 @@ public class AppConfiguration {
22 @Value("${app.zk.switch}") 22 @Value("${app.zk.switch}")
23 private boolean zkSwitch; 23 private boolean zkSwitch;
24 24
25 - @Value("${netty.server.port}")  
26 - private int port; 25 + @Value("${cim.server.port}")
  26 + private int cimServerPort;
27 27
28 - public int getPort() {  
29 - return port; 28 + @Value("${cim.clear.route.request.url}")
  29 + private String clearRouteUrl ;
  30 +
  31 + public String getClearRouteUrl() {
  32 + return clearRouteUrl;
30 } 33 }
31 34
32 - public void setPort(int port) {  
33 - this.port = port; 35 + public void setClearRouteUrl(String clearRouteUrl) {
  36 + this.clearRouteUrl = clearRouteUrl;
34 } 37 }
35 38
36 public String getZkRoot() { 39 public String getZkRoot() {
@@ -56,4 +59,12 @@ public class AppConfiguration { @@ -56,4 +59,12 @@ public class AppConfiguration {
56 public void setZkSwitch(boolean zkSwitch) { 59 public void setZkSwitch(boolean zkSwitch) {
57 this.zkSwitch = zkSwitch; 60 this.zkSwitch = zkSwitch;
58 } 61 }
  62 +
  63 + public int getCimServerPort() {
  64 + return cimServerPort;
  65 + }
  66 +
  67 + public void setCimServerPort(int cimServerPort) {
  68 + this.cimServerPort = cimServerPort;
  69 + }
59 } 70 }
1 package com.crossoverjie.cim.server.config; 1 package com.crossoverjie.cim.server.config;
2 2
  3 +import okhttp3.OkHttpClient;
3 import org.I0Itec.zkclient.ZkClient; 4 import org.I0Itec.zkclient.ZkClient;
4 import org.springframework.beans.factory.annotation.Autowired; 5 import org.springframework.beans.factory.annotation.Autowired;
5 import org.springframework.context.annotation.Bean; 6 import org.springframework.context.annotation.Bean;
6 import org.springframework.context.annotation.Configuration; 7 import org.springframework.context.annotation.Configuration;
7 8
  9 +import java.util.concurrent.TimeUnit;
  10 +
8 /** 11 /**
9 * Function: 12 * Function:
10 * 13 *
@@ -22,4 +25,18 @@ public class BeanConfig { @@ -22,4 +25,18 @@ public class BeanConfig {
22 public ZkClient buildZKClient(){ 25 public ZkClient buildZKClient(){
23 return new ZkClient(appConfiguration.getZkAddr(), 5000); 26 return new ZkClient(appConfiguration.getZkAddr(), 5000);
24 } 27 }
  28 +
  29 + /**
  30 + * http client
  31 + * @return okHttp
  32 + */
  33 + @Bean
  34 + public OkHttpClient okHttpClient() {
  35 + OkHttpClient.Builder builder = new OkHttpClient.Builder();
  36 + builder.connectTimeout(30, TimeUnit.SECONDS)
  37 + .readTimeout(10, TimeUnit.SECONDS)
  38 + .writeTimeout(10,TimeUnit.SECONDS)
  39 + .retryOnConnectionFailure(true);
  40 + return builder.build();
  41 + }
25 } 42 }
@@ -27,7 +27,7 @@ import org.springframework.web.bind.annotation.ResponseBody; @@ -27,7 +27,7 @@ import org.springframework.web.bind.annotation.ResponseBody;
27 public class IndexController { 27 public class IndexController {
28 28
29 @Autowired 29 @Autowired
30 - private CIMServer heartbeatClient ; 30 + private CIMServer cimServer;
31 31
32 32
33 /** 33 /**
@@ -46,7 +46,7 @@ public class IndexController { @@ -46,7 +46,7 @@ public class IndexController {
46 @ResponseBody 46 @ResponseBody
47 public BaseResponse<SendMsgResVO> sendMsg(@RequestBody SendMsgReqVO sendMsgReqVO){ 47 public BaseResponse<SendMsgResVO> sendMsg(@RequestBody SendMsgReqVO sendMsgReqVO){
48 BaseResponse<SendMsgResVO> res = new BaseResponse(); 48 BaseResponse<SendMsgResVO> res = new BaseResponse();
49 - heartbeatClient.sendGoogleProtoMsg(sendMsgReqVO) ; 49 + cimServer.sendMsg(sendMsgReqVO) ;
50 50
51 counterService.increment(Constants.COUNTER_SERVER_PUSH_COUNT); 51 counterService.increment(Constants.COUNTER_SERVER_PUSH_COUNT);
52 52
1 package com.crossoverjie.cim.server.handle; 1 package com.crossoverjie.cim.server.handle;
2 2
  3 +import com.alibaba.fastjson.JSONObject;
3 import com.crossoverjie.cim.common.constant.Constants; 4 import com.crossoverjie.cim.common.constant.Constants;
4 import com.crossoverjie.cim.common.protocol.CIMRequestProto; 5 import com.crossoverjie.cim.common.protocol.CIMRequestProto;
  6 +import com.crossoverjie.cim.server.config.AppConfiguration;
  7 +import com.crossoverjie.cim.server.kit.CIMUserInfo;
5 import com.crossoverjie.cim.server.util.SessionSocketHolder; 8 import com.crossoverjie.cim.server.util.SessionSocketHolder;
  9 +import com.crossoverjie.cim.server.util.SpringBeanFactory;
6 import io.netty.channel.ChannelHandler; 10 import io.netty.channel.ChannelHandler;
7 import io.netty.channel.ChannelHandlerContext; 11 import io.netty.channel.ChannelHandlerContext;
8 import io.netty.channel.SimpleChannelInboundHandler; 12 import io.netty.channel.SimpleChannelInboundHandler;
9 import io.netty.channel.socket.nio.NioSocketChannel; 13 import io.netty.channel.socket.nio.NioSocketChannel;
  14 +import okhttp3.*;
10 import org.slf4j.Logger; 15 import org.slf4j.Logger;
11 import org.slf4j.LoggerFactory; 16 import org.slf4j.LoggerFactory;
12 17
  18 +import java.io.IOException;
  19 +
13 /** 20 /**
14 * Function: 21 * Function:
15 * 22 *
@@ -22,7 +29,7 @@ public class CIMServerHandle extends SimpleChannelInboundHandler<CIMRequestProto @@ -22,7 +29,7 @@ public class CIMServerHandle extends SimpleChannelInboundHandler<CIMRequestProto
22 29
23 private final static Logger LOGGER = LoggerFactory.getLogger(CIMServerHandle.class); 30 private final static Logger LOGGER = LoggerFactory.getLogger(CIMServerHandle.class);
24 31
25 - 32 + private final MediaType mediaType = MediaType.parse("application/json");
26 /** 33 /**
27 * 取消绑定 34 * 取消绑定
28 * @param ctx 35 * @param ctx
@@ -30,9 +37,35 @@ public class CIMServerHandle extends SimpleChannelInboundHandler<CIMRequestProto @@ -30,9 +37,35 @@ public class CIMServerHandle extends SimpleChannelInboundHandler<CIMRequestProto
30 */ 37 */
31 @Override 38 @Override
32 public void channelInactive(ChannelHandlerContext ctx) throws Exception { 39 public void channelInactive(ChannelHandlerContext ctx) throws Exception {
33 - String userName = SessionSocketHolder.getUserName((NioSocketChannel) ctx.channel());  
34 - LOGGER.info("用户[{}]断开",userName); 40 + CIMUserInfo userInfo = SessionSocketHolder.getUserId((NioSocketChannel) ctx.channel());
  41 + LOGGER.info("用户[{}]下线",userInfo.getUserName());
35 SessionSocketHolder.remove((NioSocketChannel) ctx.channel()); 42 SessionSocketHolder.remove((NioSocketChannel) ctx.channel());
  43 +
  44 + //清除路由关系
  45 + clearRouteInfo(userInfo);
  46 + }
  47 +
  48 + /**
  49 + * 清除路由关系
  50 + * @param userInfo
  51 + * @throws IOException
  52 + */
  53 + private void clearRouteInfo(CIMUserInfo userInfo) throws IOException {
  54 + OkHttpClient okHttpClient = SpringBeanFactory.getBean(OkHttpClient.class);
  55 + AppConfiguration configuration = SpringBeanFactory.getBean(AppConfiguration.class);
  56 + JSONObject jsonObject = new JSONObject();
  57 + jsonObject.put("userId", userInfo.getUserId());
  58 + RequestBody requestBody = RequestBody.create(mediaType, jsonObject.toString());
  59 +
  60 + Request request = new Request.Builder()
  61 + .url(configuration.getClearRouteUrl())
  62 + .post(requestBody)
  63 + .build();
  64 +
  65 + Response response = okHttpClient.newCall(request).execute();
  66 + if (!response.isSuccessful()) {
  67 + throw new IOException("Unexpected code " + response);
  68 + }
36 } 69 }
37 70
38 71
  1 +package com.crossoverjie.cim.server.kit;
  2 +
  3 +/**
  4 + * Function:
  5 + *
  6 + * @author crossoverJie
  7 + * Date: 2018/12/24 02:33
  8 + * @since JDK 1.8
  9 + */
  10 +public class CIMUserInfo {
  11 + private Long userId ;
  12 + private String userName ;
  13 +
  14 + public CIMUserInfo(Long userId, String userName) {
  15 + this.userId = userId;
  16 + this.userName = userName;
  17 + }
  18 +
  19 + public Long getUserId() {
  20 + return userId;
  21 + }
  22 +
  23 + public void setUserId(Long userId) {
  24 + this.userId = userId;
  25 + }
  26 +
  27 + public String getUserName() {
  28 + return userName;
  29 + }
  30 +
  31 + public void setUserName(String userName) {
  32 + this.userName = userName;
  33 + }
  34 +
  35 + @Override
  36 + public String toString() {
  37 + return "CIMUserInfo{" +
  38 + "userId=" + userId +
  39 + ", userName='" + userName + '\'' +
  40 + '}';
  41 + }
  42 +}
@@ -21,11 +21,13 @@ public class RegistryZK implements Runnable { @@ -21,11 +21,13 @@ public class RegistryZK implements Runnable {
21 private AppConfiguration appConfiguration ; 21 private AppConfiguration appConfiguration ;
22 22
23 private String ip; 23 private String ip;
24 - private int port; 24 + private int cimServerPort;
  25 + private int httpPort;
25 26
26 - public RegistryZK(String ip, int port) { 27 + public RegistryZK(String ip, int cimServerPort,int httpPort) {
27 this.ip = ip; 28 this.ip = ip;
28 - this.port = port; 29 + this.cimServerPort = cimServerPort;
  30 + this.httpPort = httpPort ;
29 zKit = SpringBeanFactory.getBean(ZKit.class) ; 31 zKit = SpringBeanFactory.getBean(ZKit.class) ;
30 appConfiguration = SpringBeanFactory.getBean(AppConfiguration.class) ; 32 appConfiguration = SpringBeanFactory.getBean(AppConfiguration.class) ;
31 } 33 }
@@ -38,7 +40,7 @@ public class RegistryZK implements Runnable { @@ -38,7 +40,7 @@ public class RegistryZK implements Runnable {
38 40
39 //是否要将自己注册到 ZK 41 //是否要将自己注册到 ZK
40 if (appConfiguration.isZkSwitch()){ 42 if (appConfiguration.isZkSwitch()){
41 - String path = appConfiguration.getZkRoot() + "/ip-" + ip + ":" + port; 43 + String path = appConfiguration.getZkRoot() + "/ip-" + ip + ":" + cimServerPort + ":" + httpPort;
42 zKit.createNode(path, path); 44 zKit.createNode(path, path);
43 logger.info("注册 zookeeper 成功，msg=[{}]", path); 45 logger.info("注册 zookeeper 成功，msg=[{}]", path);
44 } 46 }
1 package com.crossoverjie.cim.server.server; 1 package com.crossoverjie.cim.server.server;
2 2
3 import com.alibaba.fastjson.JSON; 3 import com.alibaba.fastjson.JSON;
  4 +import com.crossoverjie.cim.common.constant.Constants;
4 import com.crossoverjie.cim.common.pojo.CustomProtocol; 5 import com.crossoverjie.cim.common.pojo.CustomProtocol;
5 import com.crossoverjie.cim.common.protocol.CIMRequestProto; 6 import com.crossoverjie.cim.common.protocol.CIMRequestProto;
6 import com.crossoverjie.cim.server.init.CIMServerInitializer; 7 import com.crossoverjie.cim.server.init.CIMServerInitializer;
@@ -41,7 +42,7 @@ public class CIMServer { @@ -41,7 +42,7 @@ public class CIMServer {
41 private EventLoopGroup work = new NioEventLoopGroup(); 42 private EventLoopGroup work = new NioEventLoopGroup();
42 43
43 44
44 - @Value("${netty.server.port}") 45 + @Value("${cim.server.port}")
45 private int nettyPort; 46 private int nettyPort;
46 47
47 48
@@ -101,15 +102,16 @@ public class CIMServer { @@ -101,15 +102,16 @@ public class CIMServer {
101 * 发送 Google Protocol 编码消息 102 * 发送 Google Protocol 编码消息
102 * @param sendMsgReqVO 消息 103 * @param sendMsgReqVO 消息
103 */ 104 */
104 - public void sendGoogleProtoMsg(SendMsgReqVO sendMsgReqVO){  
105 - NioSocketChannel socketChannel = SessionSocketHolder.get(sendMsgReqVO.getId()); 105 + public void sendMsg(SendMsgReqVO sendMsgReqVO){
  106 + NioSocketChannel socketChannel = SessionSocketHolder.get(sendMsgReqVO.getUserId());
106 107
107 if (null == socketChannel) { 108 if (null == socketChannel) {
108 - throw new NullPointerException("没有[" + sendMsgReqVO.getId() + "]的socketChannel"); 109 + throw new NullPointerException("客户端[" + sendMsgReqVO.getUserId() + "]不在线！");
109 } 110 }
110 CIMRequestProto.CIMReqProtocol protocol = CIMRequestProto.CIMReqProtocol.newBuilder() 111 CIMRequestProto.CIMReqProtocol protocol = CIMRequestProto.CIMReqProtocol.newBuilder()
111 - .setRequestId((int) sendMsgReqVO.getId()) 112 + .setRequestId(sendMsgReqVO.getUserId())
112 .setReqMsg(sendMsgReqVO.getMsg()) 113 .setReqMsg(sendMsgReqVO.getMsg())
  114 + .setType(Constants.CommandType.MSG)
113 .build(); 115 .build();
114 116
115 ChannelFuture future = socketChannel.writeAndFlush(protocol); 117 ChannelFuture future = socketChannel.writeAndFlush(protocol);
1 package com.crossoverjie.cim.server.util; 1 package com.crossoverjie.cim.server.util;
2 2
  3 +import com.crossoverjie.cim.server.kit.CIMUserInfo;
3 import io.netty.channel.socket.nio.NioSocketChannel; 4 import io.netty.channel.socket.nio.NioSocketChannel;
4 5
5 import java.util.Map; 6 import java.util.Map;
@@ -41,17 +42,20 @@ public class SessionSocketHolder { @@ -41,17 +42,20 @@ public class SessionSocketHolder {
41 * @param nioSocketChannel 42 * @param nioSocketChannel
42 * @return 43 * @return
43 */ 44 */
44 - public static String getUserName(NioSocketChannel nioSocketChannel){ 45 + public static CIMUserInfo getUserId(NioSocketChannel nioSocketChannel){
45 for (Map.Entry<Long, NioSocketChannel> entry : CHANNEL_MAP.entrySet()) { 46 for (Map.Entry<Long, NioSocketChannel> entry : CHANNEL_MAP.entrySet()) {
46 NioSocketChannel value = entry.getValue(); 47 NioSocketChannel value = entry.getValue();
47 if (nioSocketChannel == value){ 48 if (nioSocketChannel == value){
48 Long key = entry.getKey(); 49 Long key = entry.getKey();
49 String userName = SESSION_MAP.get(key); 50 String userName = SESSION_MAP.get(key);
50 - return userName ; 51 + CIMUserInfo info = new CIMUserInfo(key,userName) ;
  52 + return info ;
51 } 53 }
52 } 54 }
53 55
54 return null; 56 return null;
55 } 57 }
56 58
  59 +
  60 +
57 } 61 }
@@ -18,9 +18,9 @@ public class SendMsgReqVO extends BaseRequest { @@ -18,9 +18,9 @@ public class SendMsgReqVO extends BaseRequest {
18 @ApiModelProperty(required = true, value = "msg", example = "hello") 18 @ApiModelProperty(required = true, value = "msg", example = "hello")
19 private String msg ; 19 private String msg ;
20 20
21 - @NotNull(message = "id 不能为空")  
22 - @ApiModelProperty(required = true, value = "id", example = "11")  
23 - private long id ; 21 + @NotNull(message = "userId 不能为空")
  22 + @ApiModelProperty(required = true, value = "userId", example = "11")
  23 + private Long userId ;
24 24
25 public String getMsg() { 25 public String getMsg() {
26 return msg; 26 return msg;
@@ -30,11 +30,19 @@ public class SendMsgReqVO extends BaseRequest { @@ -30,11 +30,19 @@ public class SendMsgReqVO extends BaseRequest {
30 this.msg = msg; 30 this.msg = msg;
31 } 31 }
32 32
33 - public long getId() {  
34 - return id; 33 + public long getUserId() {
  34 + return userId;
35 } 35 }
36 36
37 - public void setId(long id) {  
38 - this.id = id; 37 + public void setUserId(long userId) {
  38 + this.userId = userId;
  39 + }
  40 +
  41 + @Override
  42 + public String toString() {
  43 + return "SendMsgReqVO{" +
  44 + "msg='" + msg + '\'' +
  45 + ", userId=" + userId +
  46 + "} " + super.toString();
39 } 47 }
40 } 48 }
1 -spring.application.name=netty-heartbeat-server 1 +spring.application.name=cim-server
2 2
3 # web port 3 # web port
4 server.port=8081 4 server.port=8081
@@ -6,7 +6,8 @@ server.port=8081 @@ -6,7 +6,8 @@ server.port=8081
6 # 是否打开swagger 6 # 是否打开swagger
7 swagger.enable = true 7 swagger.enable = true
8 8
9 -netty.server.port=11211 9 +# cim 服务器端口
  10 +cim.server.port=11211
10 11
11 logging.level.root=info 12 logging.level.root=info
12 13
@@ -28,3 +29,6 @@ app.zk.addr=47.98.194.60:2181 @@ -28,3 +29,6 @@ app.zk.addr=47.98.194.60:2181
28 29
29 # zk 注册根节点 30 # zk 注册根节点
30 app.zk.root=/route 31 app.zk.root=/route
  32 +
  33 +# 清除路由信息
  34 +cim.clear.route.request.url=http://localhost:8083/offLine
1 spring.application.name=cim-zk 1 spring.application.name=cim-zk
2 2
3 -# web port  
4 -server.port=8083 3 +# web cimServerPort
  4 +server.cimServerPort=8083
5 5
6 # 是否打开swagger 6 # 是否打开swagger
7 swagger.enable = true 7 swagger.enable = true
1 1
2 spring.application.name=spring-boot-admin 2 spring.application.name=spring-boot-admin
3 -server.port = 8888 3 +server.cimServerPort = 8888
4 4
5 logging.level.root=info 5 logging.level.root=info