dubbo自定义异常调用链路传递, netty消息同步改造

im-client/src/main/java/net/sopod/soim/client/cmd/CmdEnum.java

@@ -31,6 +31,8 @@ public enum CmdEnum {
 
     friends(FriendsHandler.class),
 
+    me(MeHandler.class),
+
     help,
 
     /** 退出 */

im-client/src/main/java/net/sopod/soim/client/handler/cmd/LoginHandler.java

@@ -50,7 +50,7 @@ public class LoginHandler implements CmdHandler<ArgsLogin> {
                 .setUid(loginRes.getUid())
                 .setToken(loginRes.getAuthToken())
                 .build();
-        soImSession.connect(clientConfig.getHost(), clientConfig.getPort(), reqTokenAuth);
+        soImSession.connect(clientConfig.getHost(), clientConfig.getPort(), reqTokenAuth, args.getAccount());
     }
 
 }

im-client/src/main/java/net/sopod/soim/client/handler/cmd/MeHandler.java

@@ -0,0 +1,28 @@
+package net.sopod.soim.client.handler.cmd;
+
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+import net.sopod.soim.client.cmd.handler.NonArgsHandler;
+import net.sopod.soim.client.logger.Logger;
+import net.sopod.soim.client.session.SoImSession;
+
+/**
+ * MeHandler
+ *
+ * @author tmy
+ * @date 2022-06-02 15:49
+ */
+@Singleton
+public class MeHandler extends NonArgsHandler {
+
+    @Inject
+    private SoImSession soImSession;
+
+    @Override
+    public void handle() {
+        Long uid = soImSession.getUid();
+        String account = soImSession.getAccount();
+        Logger.info("me: " + uid + "|" + account);
+    }
+
+}

im-client/src/main/java/net/sopod/soim/client/protocol/ImMessageInboundHandler.java

@@ -0,0 +1,22 @@
+package net.sopod.soim.client.protocol;
+
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.SimpleChannelInboundHandler;
+import net.sopod.soim.data.serialize.ImMessage;
+
+/**
+ * ImMessageInboundHandler
+ *
+ * @author tmy
+ * @date 2022-06-02 17:51
+ */
+public class ImMessageInboundHandler extends SimpleChannelInboundHandler<ImMessage> {
+
+    @Override
+    protected void channelRead0(ChannelHandlerContext channelHandlerContext, ImMessage imMessage) throws Exception {
+        // 请求序列号，complete 对应 CompletableFuture
+        int serialNo = imMessage.getSerialNo();
+
+    }
+
+}

im-client/src/main/java/net/sopod/soim/client/protocol/ImMessageOutboundHandler.java

@@ -0,0 +1,28 @@
+package net.sopod.soim.client.protocol;
+
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelOutboundHandlerAdapter;
+import io.netty.channel.ChannelPromise;
+import net.sopod.soim.data.serialize.ImMessage;
+
+/**
+ * ImMessageOutboundHandler
+ *
+ * @author tmy
+ * @date 2022-06-02 17:53
+ */
+public class ImMessageOutboundHandler extends ChannelOutboundHandlerAdapter {
+
+    @Override
+    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
+        if (!(msg instanceof ImMessage)) {
+            return;
+        }
+        ImMessage imMessage = (ImMessage) msg;
+        imMessage.setSerialNo(10086);
+        // queue.add something..., 调用的地方，通过阻塞方式 FastThreadLocal 获取一个 CompletableFuture
+
+        super.write(ctx, msg, promise);
+    }
+
+}

im-client/src/main/java/net/sopod/soim/client/protocol/ImMessageRegistry.java

@@ -0,0 +1,10 @@
+package net.sopod.soim.client.protocol;
+
+/**
+ * ImMessageRegistry
+ *
+ * @author tmy
+ * @date 2022-06-02 17:32
+ */
+public class ImMessageRegistry {
+}

im-client/src/main/java/net/sopod/soim/client/protocol/MessageQueueHolder.java

@@ -0,0 +1,18 @@
+package net.sopod.soim.client.protocol;
+
+/**
+ * MessageQueueHolder
+ * 发送一条消息产生一个序列号，在队列中等待响应消息
+ *
+ * @author tmy
+ * @date 2022-06-02 17:29
+ */
+public class MessageQueueHolder {
+
+
+
+    public void a() {
+
+    }
+
+}

im-client/src/main/java/net/sopod/soim/client/session/SoImSession.java

@@ -10,12 +10,14 @@ import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.socket.SocketChannel;
 import io.netty.channel.socket.nio.NioSocketChannel;
 import net.sopod.soim.client.logger.Logger;
+import net.sopod.soim.common.res.R;
 import net.sopod.soim.common.util.netty.Varint32FrameCodec;
 import net.sopod.soim.data.serialize.ImMessageCodec;
 import net.sopod.soim.data.msg.auth.Auth;
 import net.sopod.soim.data.msg.chat.Chat;
 import org.slf4j.LoggerFactory;
 
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
@@ -39,12 +41,14 @@ public class SoImSession {
 
     private Long uid;
 
+    private String account;
+
     @Inject
     public SoImSession(MessageDispatcher messageDispatcher) {
         this.messageDispatcher = messageDispatcher;
     }
 
-    public void connect(String host, Integer port, Auth.ReqTokenAuth tokenAuth) {
+    public void connect(String host, Integer port, Auth.ReqTokenAuth tokenAuth, String account) {
         // 关闭旧连接
         this.close();
 
@@ -57,7 +61,9 @@ public class SoImSession {
                     protected void initChannel(SocketChannel ch) throws Exception {
                         ch.pipeline()
                             .addLast(new Varint32FrameCodec())
-                            .addLast(new ImMessageCodec())
+                            .addLast(new ImMessageCodec.ProtoMsg2ImMessageEncoder())
+                            .addLast(new ImMessageCodec.ImMessage2ByteEncoder())
+                            .addLast(new ImMessageCodec.ImMessageDecoder())
                             .addLast(messageDispatcher);
                     }
                 });
@@ -66,6 +72,7 @@ public class SoImSession {
             clientChannel = b.connect(host, port).await().channel();
             // 连接后立即发送认证消息，10s未认证连接关闭
             clientChannel.writeAndFlush(tokenAuth);
+            this.account = account;
         } catch (Exception e) {
             Logger.error("连接服务器失败: {}", e.getMessage());
         }
@@ -88,17 +95,18 @@ public class SoImSession {
     /**
      * 发送消息
      */
-    public void send(MessageLite message) {
+    public <T> CompletableFuture<T> send(MessageLite message) {
         if (!auth.get()) {
             Logger.error("请先登录");
-            return;
+            return null;
         }
         if (clientChannel == null
             || !clientChannel.isActive()) {
             Logger.error("连接已关闭");
-            return;
+            return null;
         }
         clientChannel.writeAndFlush(message);
+        return new CompletableFuture<>();
     }
 
     public void textChat(String receiverName, String message) {
@@ -114,6 +122,14 @@ public class SoImSession {
         this.send(textChat);
     }
 
+    public Long getUid() {
+        return uid;
+    }
+
+    public String getAccount() {
+        return account;
+    }
+
     /**
      * 关闭 tcp 连接
      */

im-common/pom.xml

@@ -42,6 +42,12 @@
             <version>${netty.version}</version>
             <scope>provided</scope>
         </dependency>
+        <dependency>
+            <groupId>org.apache.dubbo</groupId>
+            <artifactId>dubbo-rpc-api</artifactId>
+            <version>${dubbo.version}</version>
+            <scope>provided</scope>
+        </dependency>
     </dependencies>
 
 </project>

im-common/src/main/java/net/sopod/soim/common/dubbo/SoimExceptionFilter.java

@@ -0,0 +1,32 @@
+package net.sopod.soim.common.dubbo;
+
+import net.sopod.soim.common.dubbo.exception.SoimException;
+import org.apache.dubbo.rpc.Invocation;
+import org.apache.dubbo.rpc.Invoker;
+import org.apache.dubbo.rpc.Result;
+import org.apache.dubbo.rpc.filter.ExceptionFilter;
+import org.apache.dubbo.rpc.service.GenericService;
+
+/**
+ * SoimExceptionFilter
+ *
+ * @author tmy
+ * @date 2022-06-02 11:08
+ */
+public class SoimExceptionFilter extends ExceptionFilter {
+
+    private static final String exceptPack = SoimException.class.getPackageName();
+
+    @Override
+    public void onResponse(Result appResponse, Invoker<?> invoker, Invocation invocation) {
+        if (appResponse.hasException() && GenericService.class != invoker.getInterface()) {
+            // 是自定义异常直接返回
+            if (appResponse.getException().getClass().getName()
+                    .startsWith(exceptPack)) {
+                return;
+            }
+        }
+        super.onResponse(appResponse, invoker, invocation);
+    }
+
+}

im-common/src/main/java/net/sopod/soim/common/dubbo/exception/DasException.java

@@ -0,0 +1,33 @@
+package net.sopod.soim.common.dubbo.exception;
+
+/**
+ * DasException
+ *
+ * @author tmy
+ * @date 2022-06-02 12:02
+ */
+public class DasException extends SoimException {
+
+    private static final long serialVersionUID = -533031075736547194L;
+
+    public DasException() {
+        super();
+    }
+
+    public DasException(String message) {
+        super(message);
+    }
+
+    public DasException(String message, Throwable cause) {
+        super(message, cause);
+    }
+
+    public DasException(Throwable cause) {
+        super(cause);
+    }
+
+    public DasException(String message, Throwable cause, boolean writableStackTrace) {
+        super(message, cause, writableStackTrace);
+    }
+
+}

im-common/src/main/java/net/sopod/soim/common/dubbo/exception/LogicException.java

@@ -0,0 +1,33 @@
+package net.sopod.soim.common.dubbo.exception;
+
+/**
+ * LogicException
+ *
+ * @author tmy
+ * @date 2022-06-02 12:03
+ */
+public class LogicException extends SoimException {
+
+    private static final long serialVersionUID = 8123008638375992724L;
+
+    public LogicException() {
+        super();
+    }
+
+    public LogicException(String message) {
+        super(message);
+    }
+
+    public LogicException(String message, Throwable cause) {
+        super(message, cause);
+    }
+
+    public LogicException(Throwable cause) {
+        super(cause);
+    }
+
+    public LogicException(String message, Throwable cause, boolean writableStackTrace) {
+        super(message, cause, writableStackTrace);
+    }
+
+}

im-common/src/main/java/net/sopod/soim/common/dubbo/exception/ServiceException.java

@@ -0,0 +1,34 @@
+package net.sopod.soim.common.dubbo.exception;
+
+/**
+ * 通用服务异常
+ * 默认不生成堆栈信息，设置 writableStackTrace 为 true 则生成堆栈信息
+ *
+ * @author 晚星
+ * @date 2019/12/30 16:30
+ */
+public class ServiceException extends SoimException {
+
+    private static final long serialVersionUID = 6555896535065406344L;
+
+    public ServiceException() {
+        super();
+    }
+
+    public ServiceException(String message) {
+        super(message);
+    }
+
+    public ServiceException(String message, Throwable cause) {
+        super(message, cause);
+    }
+
+    public ServiceException(Throwable cause) {
+        super(cause);
+    }
+
+    public ServiceException(String message, Throwable cause, boolean writableStackTrace) {
+        super(message, cause, writableStackTrace);
+    }
+
+}

im-common/src/main/java/net/sopod/soim/common/dubbo/exception/SoimException.java

@@ -0,0 +1,39 @@
+package net.sopod.soim.common.dubbo.exception;
+
+import java.io.Serializable;
+
+/**
+ * SoimException
+ *
+ * @author tmy
+ * @date 2022-06-02 11:29
+ */
+public class SoimException extends RuntimeException implements Serializable {
+
+    private static final long serialVersionUID = 7148131411144540939L;
+
+    public SoimException() {}
+
+    public SoimException(String message) {
+        super(message, null, false, false);
+    }
+
+    public SoimException(String message, Throwable cause) {
+        super(message, cause, false, false);
+    }
+
+    public SoimException(Throwable cause) {
+        super(cause == null ? null : cause.getMessage(), cause, false, false);
+    }
+
+    /**
+     *
+     * @param message 错误信息
+     * @param cause 错误对象
+     * @param writableStackTrace 是否启用堆栈追踪
+     */
+    public SoimException(String message, Throwable cause , boolean writableStackTrace) {
+        super(message, cause, false, writableStackTrace);
+    }
+
+}

im-common/src/main/java/net/sopod/soim/common/res/R.java

@@ -0,0 +1,11 @@
+package net.sopod.soim.common.res;
+
+/**
+ * R
+ *
+ * @author tmy
+ * @date 2022-06-02 15:31
+ */
+public class R {
+
+}

im-common/src/main/resources/META-INF/dubbo/org.apache.dubbo.rpc.Filter

@@ -0,0 +1 @@
+soimExceptionFilter=net.sopod.soim.common.dubbo.SoimExceptionFilter

im-das-api/im-das-user-api/src/main/java/net/sopod/soim/das/user/api/config/ChatMQAutoConfiguration.java

@@ -0,0 +1,16 @@
+package net.sopod.soim.das.user.api.config;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.context.annotation.Import;
+
+/**
+ * ChatRabbitMQAutoConfiguration
+ *
+ * @author tmy
+ * @date 2022-06-02 10:01
+ */
+@ConditionalOnClass(name = "org.springframework.amqp.core.Queue")
+@Import(ChatPersistentRabbitMQConfiguration.class)
+public class ChatMQAutoConfiguration {
+
+}

im-das-api/im-das-user-api/src/main/java/net/sopod/soim/das/user/api/config/ChatPersistentRabbitMQConfiguration.java

@@ -8,6 +8,7 @@ import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.support.converter.MessageConverter;
 import org.springframework.beans.factory.support.BeanDefinitionRegistry;
 import org.springframework.beans.factory.support.RootBeanDefinition;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
@@ -15,15 +16,14 @@ import org.springframework.context.annotation.ImportBeanDefinitionRegistrar;
 import org.springframework.core.type.AnnotationMetadata;
 
 /**
- * ChatRabbitMQAutoConfiguration
+ * ChatPersistentRabbitMQConfiguration
  * RabbitMQ的配置类，用来配置队列、交换器、路由等高级信息
  *
  * @author tmy
  * @date 2022-05-30 23:14
  */
-@Import(ChatRabbitMQAutoConfiguration.class)
 @Configuration
-public class ChatRabbitMQAutoConfiguration implements ImportBeanDefinitionRegistrar {
+public class ChatPersistentRabbitMQConfiguration implements ImportBeanDefinitionRegistrar {
 
     /**
      * direct直连模式, 按照routingkey分发到指定队列

im-das-api/im-das-user-api/src/main/resources/META-INF/spring.factories

@@ -1 +1 @@
-org.springframework.boot.autoconfigure.EnableAutoConfiguration=net.sopod.soim.das.user.api.config.ChatRabbitMQAutoConfiguration
+org.springframework.boot.autoconfigure.EnableAutoConfiguration=net.sopod.soim.das.user.api.config.ChatMQAutoConfiguration

im-das/im-das-user/src/main/java/net/sopod/soim/das/user/amqp/Sender.java

@@ -23,7 +23,7 @@ import java.util.concurrent.TimeUnit;
  * @author tmy
  * @date 2022-05-28 16:23
  */
-//@Component
+@Component
 @AllArgsConstructor
 public class Sender implements ApplicationListener<ApplicationReadyEvent> {
 

im-das/im-das-user/src/main/java/net/sopod/soim/das/user/service/FriendDasImpl.java

@@ -53,6 +53,9 @@ public class FriendDasImpl implements FriendDas {
         return friendMapper.insert(imFriend);
     }
 
+    /**
+     * TODO cache
+     */
     @Override
     public Long getFriendId(Long uid, Long fid) {
         LambdaQueryWrapper<ImFriend> friendQuery = new QueryWrapper<ImFriend>().lambda()

im-service-api/im-entry-protocol/src/main/java/net/sopod/soim/data/serialize/ImMessage.java

@@ -1,5 +1,6 @@
 package net.sopod.soim.data.serialize;
 
+import com.google.protobuf.MessageLite;
 import io.netty.buffer.ByteBuf;
 
 import java.util.Random;
@@ -55,6 +56,8 @@ public class ImMessage {
 
     private byte[] body;
 
+    private transient MessageLite decodeBody;
+
     /**
      * 从 bytebuf 流读取
      */
@@ -159,4 +162,13 @@ public class ImMessage {
         return MESSAGE_HEAD_LEN + body.length;
     }
 
+    public MessageLite getDecodeBody() {
+        return decodeBody;
+    }
+
+    public ImMessage setDecodeBody(MessageLite decodeBody) {
+        this.decodeBody = decodeBody;
+        return this;
+    }
+
 }

im-service-api/im-entry-protocol/src/main/java/net/sopod/soim/data/serialize/ImMessageCodec.java

@@ -6,6 +6,7 @@ import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.CombinedChannelDuplexHandler;
 import io.netty.handler.codec.ByteToMessageDecoder;
 import io.netty.handler.codec.MessageToByteEncoder;
+import io.netty.handler.codec.MessageToMessageEncoder;
 import net.sopod.soim.data.proto.ProtoMessageManager;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -22,21 +23,36 @@ import java.util.List;
  */
 public class ImMessageCodec //extends MessageToMessageCodec<ByteBuf, MessageLite> {
         extends CombinedChannelDuplexHandler<ImMessageCodec.ProtoMsgDecoder, ImMessageCodec.ProtoMsgEncoder> {
-    public static void main(String[] args) throws ClassNotFoundException {
 
-        Type superType = ImMessageCodec.class.getGenericSuperclass();
-        String typeName = superType.getTypeName();
-        int idx = typeName.indexOf('<');
-        String genericName = typeName.substring(idx + 1, typeName.length() - 1);
-        System.out.println(genericName.trim());
-        System.out.println(Arrays.toString(genericName.split(", ")));
-    }
     private static final Logger logger = LoggerFactory.getLogger(ImMessageCodec.class);
 
     public ImMessageCodec() {
          super(new ProtoMsgDecoder(), new ProtoMsgEncoder());
     }
 
+    public static class ImMessageDecoder extends ByteToMessageDecoder {
+        @Override
+        protected void decode(ChannelHandlerContext ctx, ByteBuf byteBuf, List<Object> out) throws Exception {
+            ImMessage message = ImMessage.read(byteBuf);
+            boolean isMagicError;
+            if ((isMagicError = (message == ImMessage.MAGIC_ERROR))
+                    || message == ImMessage.PROTOCOL_ERROR) {
+                logger.warn("decode im message error: {}, remote={}, closing channel.",
+                        isMagicError ? "MagicError" : "ProtocolError",
+                        ctx.channel().remoteAddress());
+                ctx.channel().close();
+                return;
+            }
+            // 解码 protobuf 消息体
+            int serviceNo = message.getServiceNo();
+            byte[] protoByte = message.getBody();
+            MessageLite protoClass = ProtoMessageManager.getProtoInstance(serviceNo);
+            MessageLite protoMsg = protoClass.getParserForType().parseFrom(protoByte);
+            message.setDecodeBody(protoMsg);
+            out.add(message);
+        }
+    }
+
     public static class ProtoMsgDecoder extends ByteToMessageDecoder {
         @Override
         protected void decode(ChannelHandlerContext ctx, ByteBuf byteBuf, List<Object> list) throws Exception {
@@ -52,6 +68,7 @@ public class ImMessageCodec //extends MessageToMessageCodec<ByteBuf, MessageLite
             }
             // 解码 protobuf 消息体
             int serviceNo = message.getServiceNo();
+            // message.getSerialNo()
             byte[] protoByte = message.getBody();
             MessageLite protoClass = ProtoMessageManager.getProtoInstance(serviceNo);
             MessageLite protoMsg = protoClass.getParserForType().parseFrom(protoByte);
@@ -59,6 +76,25 @@ public class ImMessageCodec //extends MessageToMessageCodec<ByteBuf, MessageLite
         }
     }
 
+    public static class ProtoMsg2ImMessageEncoder extends MessageToMessageEncoder<MessageLite> {
+        @Override
+        protected void encode(ChannelHandlerContext ctx, MessageLite message, List<Object> out) throws Exception {
+            Integer serialNo = ProtoMessageManager.getSerialNo(message.getClass());
+            // TODO unknow class serialNo
+            ImMessage imMessage = new ImMessage()
+                    .setServiceNo(serialNo)
+                    .setBody(message.toByteArray());
+            out.add(imMessage);
+        }
+    }
+
+    public static class ImMessage2ByteEncoder extends MessageToByteEncoder<ImMessage> {
+        @Override
+        protected void encode(ChannelHandlerContext ctx, ImMessage imMessage, ByteBuf out) throws Exception {
+            imMessage.write(out);
+        }
+    }
+
     public static class ProtoMsgEncoder extends MessageToByteEncoder<MessageLite> {
         @Override
         protected void encode(ChannelHandlerContext ctx, MessageLite message, ByteBuf byteBuf) throws Exception {

im-service-api/im-router-api/src/main/java/net/sopod/soim/router/api/service/UserRouteService.java

@@ -25,7 +25,11 @@ public interface UserRouteService {
      */
     List<UserInfo> onlineUserList(String keyword);
 
-    Boolean routeTextChat(TextChat textChat);
+    /**
+     * @param friendId 好友关系id
+     * @param textChat 聊天内容
+     */
+    Boolean routeTextChat(Long friendId, TextChat textChat);
 
     List<Boolean> isOnlineUsers(List<Long> userIds);
 

im-service/im-logic-user/src/main/java/net/sopod/soim/logic/user/service/ChatServiceImpl.java

@@ -1,5 +1,8 @@
 package net.sopod.soim.logic.user.service;
 
+import net.sopod.soim.common.dubbo.exception.LogicException;
+import net.sopod.soim.common.dubbo.exception.ServiceException;
+import net.sopod.soim.common.dubbo.exception.SoimException;
 import net.sopod.soim.das.user.api.model.entity.ImUser;
 import net.sopod.soim.das.user.api.service.FriendDas;
 import net.sopod.soim.das.user.api.service.UserDas;
@@ -9,6 +12,8 @@ import net.sopod.soim.logic.common.util.RpcContextUtil;
 import net.sopod.soim.router.api.service.UserRouteService;
 import org.apache.dubbo.config.annotation.DubboReference;
 import org.apache.dubbo.config.annotation.DubboService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.Objects;
 
@@ -21,6 +26,8 @@ import java.util.Objects;
 @DubboService
 public class ChatServiceImpl implements ChatService {
 
+    private static final Logger logger = LoggerFactory.getLogger(ChatServiceImpl.class);
+
     @DubboReference
     private UserRouteService userRouteService;
 
@@ -36,7 +43,8 @@ public class ChatServiceImpl implements ChatService {
                 || Objects.equals(textChat.getReceiverUid(), 0L)) {
             ImUser receiverUser = userDas.getNormalUserByAccount(textChat.getReceiverName());
             if (receiverUser == null) {
-                return false;
+                // TODO receiverUid 由客户端传递
+                throw new LogicException("聊天对象不存在");
             }
             textChat.setReceiverUid(receiverUser.getId());
         }
@@ -44,11 +52,11 @@ public class ChatServiceImpl implements ChatService {
         Long friendId = friendDas.getFriendId(textChat.getUid(), textChat.getReceiverUid());
         // 不是好友
         if (friendId == null) {
-            return false;
+            throw new LogicException("请添加好友后发送消息");
         }
         // 设置调用 router 为消息接受者地址
         RpcContextUtil.setContextUid(textChat.getReceiverUid());
-        return userRouteService.routeTextChat(textChat);
+        return userRouteService.routeTextChat(friendId, textChat);
     }
 
 }

im-service/im-logic-user/src/main/resources/application.yml

@@ -15,3 +15,4 @@ dubbo:
   provider:
     retries: 0
     timeout: 2000
+    filter: soimExceptionFilter,-exception

im-service/im-router/pom.xml

@@ -22,6 +22,11 @@
             <artifactId>im-entry-api</artifactId>
             <version>${soim.version}</version>
         </dependency>
+        <dependency>
+            <groupId>net.sopod</groupId>
+            <artifactId>im-segment-id-api</artifactId>
+            <version>${soim.version}</version>
+        </dependency>
         <dependency>
             <groupId>net.sopod</groupId>
             <artifactId>im-das-user-api</artifactId>

im-service/im-router/src/main/java/net/sopod/soim/router/service/UserRouteServiceImpl.java

@@ -1,12 +1,19 @@
 package net.sopod.soim.router.service;
 
 import net.sopod.soim.common.constant.DubboConstant;
+import net.sopod.soim.common.dubbo.exception.ServiceException;
 import net.sopod.soim.common.util.ImClock;
 import net.sopod.soim.common.util.StringUtil;
+import net.sopod.soim.das.user.api.config.LogicTables;
+import net.sopod.soim.das.user.api.model.entity.ImMessage;
 import net.sopod.soim.das.user.api.model.entity.ImUser;
+import net.sopod.soim.das.user.api.mq.ChatQueue;
+import net.sopod.soim.das.user.api.mq.ChatQueueType;
+import net.sopod.soim.das.user.api.service.FriendDas;
 import net.sopod.soim.das.user.api.service.UserDas;
 import net.sopod.soim.entry.api.service.OnlineUserService;
 import net.sopod.soim.entry.api.service.TextChatService;
+import net.sopod.soim.logic.api.segmentid.core.SegmentIdGenerator;
 import net.sopod.soim.logic.common.model.TextChat;
 import net.sopod.soim.logic.common.model.UserInfo;
 import net.sopod.soim.logic.common.util.RpcContextUtil;
@@ -20,7 +27,9 @@ import org.apache.dubbo.config.annotation.DubboService;
 import org.apache.dubbo.rpc.RpcContext;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
 
+import javax.annotation.Resource;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.stream.Collectors;
@@ -32,7 +41,7 @@ import java.util.stream.Stream;
  * @author tmy
  * @date 2022-04-28 9:47
  */
-@DubboService()
+@DubboService
 public class UserRouteServiceImpl implements UserRouteService {
 
     private static final Logger logger = LoggerFactory.getLogger(UserRouteServiceImpl.class);
@@ -40,12 +49,20 @@ public class UserRouteServiceImpl implements UserRouteService {
     @DubboReference
     private UserDas userDas;
 
+    @DubboReference
+    private FriendDas friendDas;
+
     @DubboReference
     private TextChatService textChatService;
 
     @DubboReference
     private OnlineUserService onlineUserService;
 
+    @Resource
+    private SegmentIdGenerator segmentIdGenerator;
+
+    @Resource
+    private RabbitTemplate rabbitTemplate;
 
     @Override
     public RegistryRes registryUserEntry(Long uid, String imEntryAddr) {
@@ -81,8 +98,16 @@ public class UserRouteServiceImpl implements UserRouteService {
      * 调用该方法时，将到 im-router 服务的路由 uid 设置为消息接受者的 uid
      */
     @Override
-    public Boolean routeTextChat(TextChat textChat) {
+    public Boolean routeTextChat(Long friendId, TextChat textChat) {
-        // TODO 消息队列存储
+        // 通过消息队列持久化存储到db
+        ImMessage imMessage = new ImMessage()
+                .setFriendId(friendId)
+                .setId(segmentIdGenerator.nextId(LogicTables.IM_MESSAGE))
+                .setContent(textChat.getMessage())
+                .setSender(textChat.getUid())
+                .setReceiver(textChat.getReceiverUid())
+                .setCreateTime(ImClock.date());
+        rabbitTemplate.convertAndSend(ChatQueueType.IM_MESSAGE.getQueueName(), imMessage);
 
         RpcContextUtil.setContextUid(textChat.getReceiverUid());
         Boolean send = textChatService.sendTextChat(textChat);

im-service/im-router/src/main/resources/application.yml

@@ -11,6 +11,11 @@ spring:
         max-idle: 100
         max-wait: 1000
         min-idle: 2
+  rabbitmq:
+    host: 124.222.131.236
+    port: 3672
+    username: soim
+    password: sopod@rabbit#
 
 dubbo:
   application:
@@ -31,3 +36,5 @@ dubbo:
     retries: 0 # 这里服务重试时会路由到非uid所在对应im-router
     timeout: 2000
     register: false # 不自动注册，数据初始化后注册
+    filter: soimExceptionFilter,-exception
+