Skip to content

W
wecloud_im_server
Commit 4282e0ce by giaogiao
Options

Merge branch '1.4-elk' of https://gitlab.aillo.cc/hewei/wecloud_im_server into feature-cluster 

 Conflicts:
	config/src/main/resources/config/logback.xml
	开发记录.md
parents cf9e15fd bc357dba
Showing with 508 additions and 310 deletions
common/src/main/java/com/wecloud/im/ws/cache/UserCache.java
package com.wecloud.im.ws.cache;
import com.wecloud.im.ws.utils.InitIp;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Service;
import java.net.InetAddress;
import java.net.NetworkInterface;
import java.net.SocketException;
import java.util.Enumeration;
/**
* @author hewei123@163.com
* @Description 用户与redis绑定
......@@ -18,118 +13,39 @@ import java.util.Enumeration;
@Slf4j
public class UserCache {
// @Autowired
// private RedisUtils redisUtils;
/**
* 在线状态
*/
public static final Integer ONLINE = 1;
/**
* 离线状态
*/
public static final Integer OFFLINE = 0;
/**
* key name
*/
private static final String KEY_BASE = "wsu:";
/**
* 用户在线状态fieldKey
*/
private static final String ONLINE_STATUS_KEY = "st";
/**
* 用户所连机器ip的fieldKey
*/
private static final String PRIVATE_IP_KEY = "lip";
// /**
// * 在线状态
// */
// public static final Integer ONLINE = 1;
// /**
// * 离线状态
// */
// public static final Integer OFFLINE = 0;
// 用户语言暂时没用到
// private static final String LANGUAGE = "la";
// 用户公网ip，在公网部署集群需要用到
// private static final String PUBLIC_IP = "iip";
/**
* 排除无效的mac地址
*/
private final static byte[][] INVALID_MACS = {
{0x00, 0x05, 0x69}, // VMWare
{0x00, 0x1C, 0x14}, // VMWare
{0x00, 0x0C, 0x29}, // VMWare
{0x00, 0x50, 0x56}, // VMWare
{0x08, 0x00, 0x27}, // Virtualbox
{0x0A, 0x00, 0x27}, // Virtualbox
{0x00, 0x03, (byte) 0xFF}, // Virtual-PC
{0x00, 0x15, 0x5D} // Hyper-V
};
/**
* 内网ip
* key name
* websocket user info
*/
private static String lAN_IP = "";
private static final String KEY_BASE = "wui:";
// 获取机器内网ip
static {
lAN_IP = getLocalIpAddress();
log.info("lAN_IP:" + lAN_IP);
}
// /**
// * 用户在线状态fieldKey
// */
// private static final String ONLINE_STATUS_KEY = "ost";
/**
* 判断是否为虚拟mac地址
*
* @param mac
* @return
* 用户所连机器ip的fieldKey
*/
public static boolean isVmMac(byte[] mac) {
if (null == mac) {
return false;
}
for (byte[] invalid : INVALID_MACS) {
if (invalid[0] == mac[0] && invalid[1] == mac[1] && invalid[2] == mac[2]) {
return true;
}
}
private static final String LAN_IP = "lip";
return false;
}
/**
* 获取本机地址
* 用户公网ip，在公网部署集群需要用到
*/
public static String getLocalIpAddress() {
try {
Enumeration<NetworkInterface> networkInterfaces = NetworkInterface.getNetworkInterfaces();
while (networkInterfaces.hasMoreElements()) {
NetworkInterface ni = networkInterfaces.nextElement();
/*
排除docker虚拟网卡
*/
String docker0 = "docker0";
if (ni.getName().equals(docker0)) {
continue;
}
if (!ni.isUp() || ni.isLoopback() || ni.isVirtual()) {
continue;
}
if (isVmMac(ni.getHardwareAddress())) {
continue;
}
Enumeration<InetAddress> inetAddresses = ni.getInetAddresses();
while (inetAddresses.hasMoreElements()) {
InetAddress inetAddress = inetAddresses.nextElement();
if (inetAddress.isLinkLocalAddress()) {
continue;
}
return inetAddress.getHostAddress();
}
}
} catch (SocketException e) {
log.info("获取本机IP地址失败。" + e);
}
return StringUtils.EMPTY;
}
private static final String PUBLIC_IP = "pip";
private static final String type = "ty";
private static final String deviceToken = "dt";
/**
......@@ -138,7 +54,7 @@ public class UserCache {
* @param id
*/
public void online(String id) {
log.info("ws用户上线保存redis连接ip:" + lAN_IP, ",uid:");
log.info("ws用户上线保存redis连接ip:" + InitIp.lAN_IP, ",uid:");
// redisUtils.hset(KEY_BASE + id, PRIVATE_IP_KEY, lAN_IP);
// redisUtils.hset(KEY_BASE + id, ONLINE_STATUS_KEY, String.valueOf(ONLINE));
}
......
common/src/main/java/com/wecloud/im/ws/model/UserSession.java 0 → 100644
package com.wecloud.im.ws.model;
import lombok.Data;
import java.io.Serializable;
@Data
public class UserSession implements Serializable {
/**
* 通道id
*/
private String channelId;
private String deviceType;
private String deviceToken;
private String token;
}
common/src/main/java/com/wecloud/im/ws/model/redis/UserConnectionInfo.java 0 → 100644
package com.wecloud.im.ws.model.redis;
import lombok.Data;
import java.io.Serializable;
/**
* 维护client多端的ip及平台类型:
*/
@Data
public class UserConnectionInfo implements Serializable {
private String pubIp;
private String lanIp;
private String type;
private String deviceToken;
}
common/src/main/java/com/wecloud/im/ws/service/MangerChannelService.java
package com.wecloud.im.ws.service;
import com.wecloud.im.ws.model.UserSession;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.AttributeKey;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
......@@ -21,6 +23,12 @@ public interface MangerChannelService {
*/
Map<String, NioSocketChannel> CHANNEL_MAP = new ConcurrentHashMap<>();
/**
* 本地维护 uid 对应 channel_shortID
*/
Map<String, List<UserSession>> USER_SESSION_MAP = new ConcurrentHashMap<>();
/**
* CLIENT_ID,是客户端的字符串id
*/
......
common/src/main/java/com/wecloud/im/ws/service/impl/MangerChannelServiceImpl.java
......@@ -66,9 +66,6 @@ public class MangerChannelServiceImpl implements MangerChannelService {
nioSocketChannel.close();
}
// AppHashValueModel appHashValueModel = new AppHashValueModel();
// appHashValueModel.setOnlineStatus(UserCache.ONLINE);
// userCache.online(userId);
MangerChannelService.CHANNEL_MAP.put(appKey + clientId, channel);
......@@ -106,176 +103,6 @@ public class MangerChannelServiceImpl implements MangerChannelService {
//
// }
//
//
// /**
// * TODO 待完成: 根据ACK回执 以及线程等待超时机制来判断客户端是否离线和超时;
// * TODO 待完成: 发送后阻塞当前子线程2秒后获取ack回执 如客户端发起ack回执则需要主动唤醒当前子线程 立马唤醒当前子线程， 判断如果已回执则返回发送成功， 如果未回执则判断客户端是否断线或发送错误
// *
// * @param msg
// * @param userId
// * @return
// */
// @Override
// public boolean rpcWriteData(String msg, Long userId) {
//
// Future<Boolean> future = THREAD_POOL_RPC_WRITE_EXECUTOR.submit(() -> {
//
// NioSocketChannel nioSocketChannel = get(String.valueOf(userId));
// if (null == nioSocketChannel) {
// userCache.offline(String.valueOf(userId));
// log.info("rpc-writeData连接为空:" + userId + "," + msg);
// return false;
// }
//
// // 判断连接是否断开
// if (nioSocketChannel.isShutdown()) {
// log.info("rpc-writeData连接断开:" + userId + "," + msg + ",\nchannelId:" + nioSocketChannel.id().asLongText());
// return false;
// }
//
// if (log.isInfoEnabled()) {
// log.info("rpc-writeData:" + userId + "," + msg + ",\nchannelId:" + nioSocketChannel.id().asLongText());
// }
//
// ChannelFuture channelFuture = nioSocketChannel.writeAndFlush(new TextWebSocketFrame(msg));
// channelFuture.addListener(
// //执行后回调的方法
// (ChannelFutureListener) channelFuture1 -> {
// if (log.isInfoEnabled()) {
// log.info("rpc-netty线程异步执行结果:" + channelFuture1.isDone() + ",业务执行结果:" + channelFuture1.isSuccess()
// + "；\nwriteData:" + userId + "," + msg + ",channelId:" + nioSocketChannel.id().asLongText());
// }
// });
//
// channelFuture.get();
//
// return true;
// });
//
// boolean resultStatus = false;
// try {
// resultStatus = future.get();
// } catch (InterruptedException | ExecutionException e) {
// e.printStackTrace();
// }
// return resultStatus;
//
// }
//
// @Override
// public boolean rpcKickWriteData(String msg, Long userId) {
// Future<Boolean> future = THREAD_POOL_RPC_WRITE_EXECUTOR.submit(() -> {
//
// NioSocketChannel nioSocketChannel = get(String.valueOf(userId));
// if (null == nioSocketChannel) {
// log.info("rpc-kickWriteData连接为空:" + userId + "," + msg);
// return false;
// }
//
// // 判断连接是否断开
// if (nioSocketChannel.isShutdown()) {
// log.info("rpc-kickWriteData连接断开:" + userId + "," + msg + ",\nchannelId:" + nioSocketChannel.id().asLongText());
// nioSocketChannel.close();
// return false;
// }
//
// if (log.isDebugEnabled()) {
// log.info("rpc-kickWriteData:" + userId + "," + msg + ",\nchannelId:" + nioSocketChannel.id().asLongText());
// }
//
// ChannelFuture channelFuture = nioSocketChannel.writeAndFlush(new TextWebSocketFrame(msg));
// channelFuture.addListener(
// //执行后回调的方法
// (ChannelFutureListener) channelFuture1 -> {
// if (log.isDebugEnabled()) {
// log.info("rpc-netty踢人线程异步执行结果:" + channelFuture1.isDone() + ",业务执行结果:" + channelFuture1.isSuccess()
// + "；\nkickWriteData:" + userId + "," + msg + ",channelId:" + nioSocketChannel.id().asLongText());
// }
// });
//
// channelFuture.get();
// // 关闭
// nioSocketChannel.close();
// return true;
// });
//
// boolean resultStatus = false;
// try {
// resultStatus = future.get();
// } catch (InterruptedException | ExecutionException e) {
// e.printStackTrace();
// }
// return resultStatus;
// }
//
// @Override
// public boolean rpcCloseOldChannel(Long userId) {
// Future<Boolean> future = THREAD_POOL_RPC_WRITE_EXECUTOR.submit(() -> {
// NioSocketChannel nioSocketChannel = get(String.valueOf(userId));
// if (null == nioSocketChannel) {
// log.info("rpc-closeOldChannel连接为空:" + userId);
// return false;
// }
// // 关闭
// nioSocketChannel.close();
// return true;
// });
// boolean resultStatus = false;
// try {
// resultStatus = future.get();
// } catch (InterruptedException | ExecutionException e) {
// e.printStackTrace();
// }
// return resultStatus;
// }
//
// @Override
// public boolean writeData(String msg, Long userId) {
//
//// Future<Boolean> future = THREAD_POOL_EXECUTOR.submit(() -> {
//
// NioSocketChannel nioSocketChannel = get(String.valueOf(userId));
// if (null == nioSocketChannel) {
// userCache.offline(String.valueOf(userId));
// log.info("writeData连接为空:" + userId + "," + msg);
// return false;
// }
//
// // 判断连接是否断开
// if (nioSocketChannel.isShutdown()) {
// log.info("writeData连接断开:" + userId + "," + msg + ",\nchannelId:" + nioSocketChannel.id().asLongText());
// return false;
// }
//
// if (log.isDebugEnabled()) {
// log.info("writeData:" + userId + "," + msg + ",\nchannelId:" + nioSocketChannel.id().asLongText());
// }
//
// ChannelFuture channelFuture = nioSocketChannel.writeAndFlush(new TextWebSocketFrame(msg));
// channelFuture.addListener(
// //执行后回调的方法
// (ChannelFutureListener) channelFuture1 -> {
// if (log.isDebugEnabled()) {
// log.info("netty线程异步执行结果:" + channelFuture1.isDone() + ",业务执行结果:" + channelFuture1.isSuccess()
// + "；\nwriteData:" + userId + "," + msg + ",channelId:" + nioSocketChannel.id().asLongText());
// }
// });
//
//// channelFuture.get();
//
// return true;
//// });
//
//// Boolean resultStatus = false;
//// try {
//// resultStatus = future.get();
//// } catch (InterruptedException | ExecutionException e) {
//// e.printStackTrace();
//// }
//// return resultStatus;
//
// }
/**
* 获取用户在线状态
*
......@@ -287,7 +114,6 @@ public class MangerChannelServiceImpl implements MangerChannelService {
public boolean getOnlineStatus(String toAppKey, String toClientId) {
NioSocketChannel nioSocketChannel = get(toAppKey, toClientId);
if (null == nioSocketChannel) {
// userCache.offline(toAppKey + toClientId);
if (log.isDebugEnabled()) {
log.info("writeData 不存在 连接为空:" + toAppKey + toClientId);
}
......@@ -309,23 +135,16 @@ public class MangerChannelServiceImpl implements MangerChannelService {
NioSocketChannel nioSocketChannel = get(toAppKey, toClientId);
if (null == nioSocketChannel) {
// userCache.offline(toAppKey + toClientId);
// if (log.isDebugEnabled()) {
log.info("writeData连接为空:" + toAppKey + toClientId + "," + msg);
// }
return false;
}
// 判断连接是否断开
if (nioSocketChannel.isShutdown()) {
// if (log.isDebugEnabled()) {
log.info("writeData连接断开:" + toAppKey + toClientId + "," + msg + ",\nchannelId:" + nioSocketChannel.id().asLongText());
// }
return false;
}
// if (log.isDebugEnabled()) {
log.info("writeData:" + toAppKey + "," + toClientId + "," + msg + ",\nchannelId:" + nioSocketChannel.id().asLongText());
// }
ChannelFuture channelFuture = nioSocketChannel.writeAndFlush(new TextWebSocketFrame(msg));
channelFuture.addListener(
......@@ -337,7 +156,6 @@ public class MangerChannelServiceImpl implements MangerChannelService {
}
});
return true;
}
}
common/src/main/java/com/wecloud/im/ws/utils/InitIp.java 0 → 100644
package com.wecloud.im.ws.utils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Service;
import java.net.InetAddress;
import java.net.NetworkInterface;
import java.net.SocketException;
import java.util.Enumeration;
@Slf4j
@Service
public class InitIp {
/**
* 排除无效的mac地址
*/
private final static byte[][] INVALID_MACS = {
{0x00, 0x05, 0x69}, // VMWare
{0x00, 0x1C, 0x14}, // VMWare
{0x00, 0x0C, 0x29}, // VMWare
{0x00, 0x50, 0x56}, // VMWare
{0x08, 0x00, 0x27}, // Virtualbox
{0x0A, 0x00, 0x27}, // Virtualbox
{0x00, 0x03, (byte) 0xFF}, // Virtual-PC
{0x00, 0x15, 0x5D} // Hyper-V
};
/**
* 内网ip
*/
public static String lAN_IP = "";
// 获取机器内网ip
static {
lAN_IP = getLocalIpAddress();
log.info("lAN_IP:" + lAN_IP);
}
/**
* 判断是否为虚拟mac地址
*
* @param mac
* @return
*/
public static boolean isVmMac(byte[] mac) {
if (null == mac) {
return false;
}
for (byte[] invalid : INVALID_MACS) {
if (invalid[0] == mac[0] && invalid[1] == mac[1] && invalid[2] == mac[2]) {
return true;
}
}
return false;
}
/**
* 获取本机地址
*/
public static String getLocalIpAddress() {
try {
Enumeration<NetworkInterface> networkInterfaces = NetworkInterface.getNetworkInterfaces();
while (networkInterfaces.hasMoreElements()) {
NetworkInterface ni = networkInterfaces.nextElement();
/*
排除docker虚拟网卡
*/
String docker0 = "docker0";
if (ni.getName().equals(docker0)) {
continue;
}
if (!ni.isUp() || ni.isLoopback() || ni.isVirtual()) {
continue;
}
if (isVmMac(ni.getHardwareAddress())) {
continue;
}
Enumeration<InetAddress> inetAddresses = ni.getInetAddresses();
while (inetAddresses.hasMoreElements()) {
InetAddress inetAddress = inetAddresses.nextElement();
if (inetAddress.isLinkLocalAddress()) {
continue;
}
return inetAddress.getHostAddress();
}
}
} catch (SocketException e) {
log.info("获取本机IP地址失败。" + e);
}
return StringUtils.EMPTY;
}
}
config/src/main/resources/config/logback.xml
......@@ -76,18 +76,90 @@
<appender-ref ref="ERROR_FILE"/>
</appender>
<!-- <appender name="LOGSTASH"-->
<!-- class="net.logstash.logback.appender.LogstashTcpSocketAppender">-->
<!-- <destination>139.9.6.183:5044</destination>-->
<!-- &lt;!&ndash; encoder必须配置,有多种可选 &ndash;&gt;-->
<!-- <encoder charset="UTF-8"-->
<!-- class="net.logstash.logback.encoder.LogstashEncoder">-->
<!-- &lt;!&ndash; "appname":"xxx" 的作用是指定创建索引的名字时用，并且在生成的文档中会多了这个字段 &ndash;&gt;-->
<!-- <customFields>{"appname":"wc_im"}</customFields>-->
<!-- </encoder>-->
<!-- </appender>-->
<appender name="ELASTIC" class="com.internetitem.logback.elasticsearch.ElasticsearchAppender">
<url>http://elastic:ZwnEAr3arDtmc8R5aFdH@139.9.6.183:9200/_bulk</url>
<index>wc-im-%date{yyyy-MM-dd}</index>
<!-- <type>test</type>-->
<loggerName>wc-im-logger</loggerName> <!-- optional -->
<errorLoggerName>wc-im-error-logger</errorLoggerName> <!-- optional -->
<connectTimeout>30000</connectTimeout> <!-- optional (in ms, default 30000) -->
<errorsToStderr>false</errorsToStderr> <!-- optional (default false) -->
<includeCallerData>false</includeCallerData> <!-- optional (default false) -->
<logsToStderr>false</logsToStderr> <!-- optional (default false) -->
<maxQueueSize>104857600</maxQueueSize> <!-- optional (default 104857600) -->
<maxRetries>3</maxRetries> <!-- optional (default 3) -->
<readTimeout>30000</readTimeout> <!-- optional (in ms, default 30000) -->
<sleepTime>250</sleepTime> <!-- optional (in ms, default 250) -->
<rawJsonMessage>false</rawJsonMessage> <!-- optional (default false) -->
<includeMdc>false</includeMdc> <!-- optional (default false) -->
<maxMessageSize>-1</maxMessageSize> <!-- optional (default -1 -->
<authentication class="com.internetitem.logback.elasticsearch.config.BasicAuthentication"/> <!-- optional -->
<properties>
<property>
<name>host</name>
<value>${HOSTNAME}</value>
<allowEmpty>false</allowEmpty>
</property>
<property>
<name>level</name>
<value>%level</value>
</property>
<property>
<name>thread</name>
<value>%thread</value>
</property>
<property>
<name>stacktrace</name>
<value>%ex</value>
</property>
<property>
<name>package</name>
<value>%logger</value>
</property>
</properties>
<headers>
<header>
<name>Content-Type</name>
<value>application/json</value>
</header>
</headers>
</appender>
<!-- 不同环境的日志级别配置 -->
<springProfile name="dev">
<logger name="io.geekidea.springbootplus" level="DEBUG"/>
</springProfile>
<!-- 解决SpringBootAdmin错误日志问题 -->
<logger name="org.apache.catalina.connector.CoyoteAdapter" level="OFF"/>
<root level="DEBUG">
<root level="INFO">
<appender-ref ref="ELASTIC"/>
<appender-ref ref="CONSOLE"/>
<appender-ref ref="ASYNC_FILE"/>
<appender-ref ref="ASYNC_ERROR_FILE"/>
</root>
<!-- <springProfile name="eslog">-->
<!-- <root level="INFO">-->
<!-- <appender-ref ref="ELASTIC"/>-->
<!-- </root>-->
<!-- </springProfile>-->
</configuration>
\ No newline at end of file
docs/md/客户/wecloud-im即时通讯聊天云服务_外部客户对接文档.md
# wecloud-im 即时通讯云对接文档
# wecloud-im 即时通讯云对接文档
......@@ -178,3 +178,23 @@ public static void main(String[] args) {
## 消息数据加密
#### 自定义加密算法
如果客户方对数据安全比较敏感, 建议对消息聊天中的字符串数据进行**RSA不对称加密**或**AES对称加密**等数据加密技术,客户方可以任意选择世界上任一种加密算法, 加密和解密都在客户方应用内进行, wecloud平台将无法得知明文数据.
例如: 客户端发送文本消息 : "给我银行卡账号汇款10万元,卡号是10101010",
客户方可以在自己系统内维护RSA或AES的密钥, 对本文中的内容进行加密 : "给我银行卡账号汇款10万元,卡号是10101010",加密后的消息数据在wecloud将不会以明文的方式存储,以保障数据安全.
业务方服务端需要保存生成的加密密钥, 否则数据将无法解密
#### wecloud提供的加密
wecloud为提供**HTTPS**,以安全为目标的 HTTP 通道，在HTTP的基础上通过传输加密和身份认证保证了传输过程的安全性,。HTTPS 在HTTP 的基础下加入**SSL**，HTTPS 的安全基础是 SSL,。保障终端设备到服务器传输时的数据安全,不被窃取.
pom.xml
......@@ -327,6 +327,18 @@
<dependencies>
<dependency>
<groupId>net.logstash.logback</groupId>
<artifactId>logstash-logback-encoder</artifactId>
<version>5.3</version>
</dependency>
<dependency>
<groupId>com.internetitem</groupId>
<artifactId>logback-elasticsearch-appender</artifactId>
<version>1.6</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
......
开发记录.md
......@@ -17,6 +17,13 @@ aaaaa3
--
# 国内测试外网api文档
https://imapitest.wecloud.cn/api//doc.html#/home
# 国内测试外网web示例
https://imwebtest.wecloud.cn/
## 会话
1邀请2
1447818154184151040
......@@ -311,5 +318,33 @@ web示例:
imwebtest.wecloud.cn
## 多端在线 + 服务器集群
本地维护 uid 对应 channel_shortID
1123:{
channelId = ababab1 ,
deviceType = ios,
deviceToken = abdsbsn,
token = sjn321sdjfsdf
}
1123:{
channelId = ababab2 ,
deviceType = android,
deviceToken = abdsbsn,
token = sjnsdj123fsdf
}
1123:{
channelId = ababab3 ,
deviceType = web,
deviceToken = abdsb123sn,
token = null
}
飞蛙测试外网
https://testws.frogsell.com/api/doc.html#/home
\ No newline at end of file
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment