OpenFire源码学习之二十五：消息回执与离线消息(下)

这一篇紧接着上面继续了。

方案二

基于redis的消息回执。主要流程分为下面几个步骤：

1）将消息暂存储与redis中，设置好消息的过期时间

2）客户端回执消息id来消灭暂存的消息

3）开通单独线程论坛在第1）步中的消息。根据消息的时间重新发送消息。如果消息第一次存放的时间大雨有效期(自定义10秒)，解析消息中的to查找用户是否还在线。如果在则T掉(因为它长时间不理会服务的重要命令)，如果不在线则将消息放置离线表。

 

OK，先来看看消息的存储格式吧。

1.MESSAGE消息 用户集合

 SADD  SOGU:[username]  [VALUE(messageID)] [VALUE(messageID)] ...

2.已读消息设备集合

 SADD  RT:[terminalid]  [VALUE(messageID)] [VALUE(messageID)] ...

3.消息内容

 HMSET  OGM:[messageID]  CREATIONDATE [VALUE]  UPDATEDATE [VALUE] STANZA [VALUE]

4.用户、设备关联

 SADD URT:[USERNAME]  [VALUE(terminalid)] .......

(先根据消息id查找时间，在java中排序后 查找stanza)

MESSAGE

--离线表

ZADD OFOFFLINE:[username]  [INDEX(时间戳)] [VALUE(messageID)] 、[VALUE]、[VALUE]......              [VALUE]

HMSET OFOFFLINE:[messageID] STANZA[VALUE]

        CREATIONDATE [VALUE]  MESSAGESIZ[VALUE]

 

将消息暂时消息存储：

public void storeMessage(String username, Packet packet) {Jedis jedis = XMPPServer.getInstance().getGroupRedisManager().getJedis();String packetID = "";if (packet instanceof Message) packetID = ((Message)packet).getID();else if (packet instanceof IQ) packetID = ((IQ)packet).getID();else return;try {jedis.sadd("SOGU:" + username, packetID);Map<String, String> hash = new HashMap<String, String>();hash.put("STANZA", packet.toXML());hash.put("CREATIONDATE", StringUtils.dateToMillis(new Date()));jedis.hmset("OGM:" + packetID, hash);} finally {XMPPServer.getInstance().getGroupRedisManager().returnRes(jedis);}htp.execute(addMessagesToDB(packet));}private Runnable addMessagesToDB(final Packet packet) {return new Runnable() {@Overridepublic void run() {MyDBopt.insertMessage(packet);}

客户端收到消息来回执服务端的操作

private void handle(IQ packet) {JID recipientJID = packet.getTo();if (IQ.Type.crs != packet.getType()) {// Check if the packet was sent to the server hostnameif (recipientJID != null && recipientJID.getNode() == null &&recipientJID.getResource() == null && serverName.equals(recipientJID.getDomain())) {Element childElement = packet.getChildElement();if (childElement != null && childElement.element("addresses") != null) {// to route this packetmulticastRouter.route(packet);return;}}}if (IQ.Type.crs == packet.getType()) {String username = packet.getFrom().getNode();String terminal = packet.getFrom().getTerminal();String msgId = packet.getID();if (username == null || msgId == null || "".equals(msgId)) {return ;}if (terminal == null) {terminal = username + "_" + System.currentTimeMillis()%1000000; }Jedis jedis = XMPPServer.getInstance().getGroupRedisManager().getJedis();try {jedis.sadd("URT:" + username, terminal);jedis.sadd("RT:" + terminal, packet.getID());} finally {XMPPServer.getInstance().getGroupRedisManager().returnRes(jedis);}threadPool.execute(createTask(msgId, username, terminal));return;}if (packet.getID() != null && (IQ.Type.result == packet.getType() || IQ.Type.error == packet.getType())) {// The server got an answer to an IQ packet that was sent from the serverIQResultListener iqResultListener = resultListeners.remove(packet.getID());if (iqResultListener != null) {resultTimeout.remove(packet.getID());if (iqResultListener != null) {try {iqResultListener.receivedAnswer(packet);}catch (Exception e) {Log.error("Error processing answer of remote entity. Answer: "+ packet.toXML(), e);}return;}}}try {// Check for registered components, services or remote serversif (recipientJID != null &&(routingTable.hasComponentRoute(recipientJID) || routingTable.hasServerRoute(recipientJID))) {// A component/service/remote server was found that can handle the PacketroutingTable.routePacket(recipientJID, packet, false);return;}if (isLocalServer(recipientJID)) {// Let the server handle the PacketElement childElement = packet.getChildElement();String namespace = null;if (childElement != null) {namespace = childElement.getNamespaceURI();}if (namespace == null) {if (packet.getType() != IQ.Type.result && packet.getType() != IQ.Type.error) {// Do nothing. We can't handle queries outside of a valid namespaceLog.warn("Unknown packet " + packet.toXML());}}else {// Check if communication to local users is allowedif (recipientJID != null && userManager.isRegisteredUser(recipientJID.getNode())) {PrivacyList list =PrivacyListManager.getInstance().getDefaultPrivacyList(recipientJID.getNode());if (list != null && list.shouldBlockPacket(packet)) {// Communication is blockedif (IQ.Type.set == packet.getType() || IQ.Type.get == packet.getType()) {// Answer that the service is unavailablesendErrorPacket(packet, PacketError.Condition.service_unavailable);}return;}}IQHandler handler = getHandler(namespace);if (handler == null) {if (recipientJID == null) {// Answer an error since the server can't handle the requested namespacesendErrorPacket(packet, PacketError.Condition.service_unavailable);}else if (recipientJID.getNode() == null ||"".equals(recipientJID.getNode())) {// Answer an error if JID is of the form <domain>sendErrorPacket(packet, PacketError.Condition.feature_not_implemented);}else {// JID is of the form <node@domain>// Answer an error since the server can't handle packets sent to a nodesendErrorPacket(packet, PacketError.Condition.service_unavailable);}}else {handler.process(packet);}}}else {// JID is of the form <node@domain/resource> or belongs to a remote server// or to an uninstalled componentroutingTable.routePacket(recipientJID, packet, false);}}catch (Exception e) {......}}

离线消息

离线消息的优化。

同样可以拓展XMPP。比如

客户端获取离线消息，可以这么通讯。

1）先向服务器询问，我总的离线消息的基本状况(有多大，有多少条)

<iq id="BfI3V-47" to="8ntmorv1ep4wgcy" type="get" from="test@8ntmorv1ep4wgcy"><query xmlns="http://jabber.org/protocol/offmsg#bif"/> </iq>

2）服务端返回

<iq type="result" id="BfI3V-47" from="8ntmorv1ep4wgcy" to="test@8ntmorv1ep4wgcy"><query xmlns="http://jabber.org/protocol/offmsg#bifs"><size>1024b</><count>128</><idset>1001,1002...</></query> </iq>

3)客户端发送分批获取命令，一次给我发10条发完为止。

<iq id="BfI3V-47" to="8ntmorv1ep4wgcy" type="get" from="test@8ntmorv1ep4wgcy"><query xmlns="http://jabber.org/protocol/offmsg#start"/><pagesize>10</> </iq>

4）服务端开始发送消息

<iq type="result" id="BfI3V-47" from="8ntmorv1ep4wgcy" to="test@8ntmorv1ep4wgcy">...... </iq> <iq type="result" id="BfI3V-47" from="8ntmorv1ep4wgcy" to="test@8ntmorv1ep4wgcy">...... </iq> .....

5)告诉客户端我都发完了

<iq type="result" id="BfI3V-47" from="8ntmorv1ep4wgcy" to="test@8ntmorv1ep4wgcy"><query xmlns="http://jabber.org/protocol/offmsg#end"/> </iq>

6)客户端本地校验，回执已经接收到的消息

<message id="BfI3V-47" to="8ntmorv1ep4wgcy" from="test1@8ntmorv1ep4wgcy/Spark 2.6.3" type="crs"/>

这里本人只是做了一个简单的示意想法。如果需要更加精准的不妨在仔细想想消息处理与格式。

 

离线消息存储。

将消息存储到redis中：

public void addMessageToRedis(Message message) {if (message == null) {return;}JID recipient = message.getTo();String username = recipient.getNode();// If the username is null (such as when an anonymous user), don't store.if (username == null || !UserManager.getInstance().isRegisteredUser(recipient)) {return;}elseif (!XMPPServer.getInstance().getServerInfo().getXMPPDomain().equals(recipient.getDomain())) {// Do not store messages sent to users of remote serversreturn;}String msgXML = message.getElement().asXML();Jedis jedis = XMPPServer.getInstance().getChatMessageJedisPoolManager().getJedis();try {String newDate = StringUtils.dateToMillis(new java.util.Date());String id = MessageIdTactics.mid(username);jedis.zadd("OFOFFLINE:" + username, Long.valueOf(newDate), id Map<String, String> hash = new HashMap<String, String>();hash.put("STANZA", msgXML);hash.put("MESSAGESIZ", String.valueOf(msgXML.length()));hash.put("CREATIONDATE", newDate);jedis.hmset("OFOFFLINE:" + id, hash);} finally {XMPPServer.getInstance().getChatMessageJedisPoolManager().returnRes(jedis);}if (sizeCache.containsKey(username)) {int size = sizeCache.get(username);size += msgXML.length();sizeCache.put(username, size);}htp.execute(addMessageToDB(message));}

Redis优化这块就到这啦。主要要做的就是：

第一：存储用户或者MUC、Group等这些都需要设置消息存储的生命周期。当用户不处于活跃状态或者长时间不登陆的。要从redis中提出。免得浪费资源。当用户重新加载的时候再将他放置redis中

第二：将需要回执消息和离线消息分开。需要回执的消息需要设置他的生命周期。离线表最好做个定时器。轮询消息。将超时出现范围内的消息(比如周期为一周)的消息同步至关系表中。这里的离线消息需要将用户的设备分开来。

这里要考虑不同的设备终端等很多不同场景，问题会比较绕口。欢迎大家和我邮件交流。

转载于:https://www.cnblogs.com/huwf/p/4273343.html

总结

以上是生活随笔为你收集整理的OpenFire源码学习之二十五：消息回执与离线消息(下)的全部内容，希望文章能够帮你解决所遇到的问题。

如果觉得生活随笔网站内容还不错，欢迎将生活随笔推荐给好友。