RabbitMQ队列配置

im-common/src/main/java/net/sopod/soim/common/util/Converter.java

@@ -0,0 +1,25 @@
+package net.sopod.soim.common.util;
+
+/**
+ * Converter
+ *
+ * @author tmy
+ * @date 2022-05-30 22:23
+ */
+public class Converter {
+
+    public static String toString(Object data) {
+        return data == null ? null : data.toString();
+    }
+
+    /**
+     * @exception  NumberFormatException  if the string cannot be parsed as an integer.
+     */
+    public static int toInt(Object value) {
+        if (value instanceof Number) {
+            return ((Number)value).intValue();
+        }
+        return Integer.parseInt(String.valueOf(value));
+    }
+
+}

im-das-api/im-das-user-api/pom.xml

@@ -28,6 +28,10 @@
             <artifactId>spring-boot-starter-amqp</artifactId>
             <scope>provided</scope>
         </dependency>
+        <dependency>
+            <groupId>org.xerial.snappy</groupId>
+            <artifactId>snappy-java</artifactId>
+        </dependency>
     </dependencies>
 
 </project>

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

@@ -0,0 +1,56 @@
+package net.sopod.soim.das.user.api.config;
+
+import net.sopod.soim.das.user.api.mq.ChatMQMessageConverter;
+import net.sopod.soim.das.user.api.mq.ChatQueueType;
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
+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.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+import org.springframework.context.annotation.ImportBeanDefinitionRegistrar;
+import org.springframework.core.type.AnnotationMetadata;
+
+/**
+ * ChatRabbitMQAutoConfiguration
+ * RabbitMQ的配置类，用来配置队列、交换器、路由等高级信息
+ *
+ * @author tmy
+ * @date 2022-05-30 23:14
+ */
+@Import(ChatRabbitMQAutoConfiguration.class)
+@Configuration
+public class ChatRabbitMQAutoConfiguration implements ImportBeanDefinitionRegistrar {
+
+    /**
+     * direct直连模式, 按照routingkey分发到指定队列
+     */
+    @Override
+    public void registerBeanDefinitions(AnnotationMetadata importingClassMetadata, BeanDefinitionRegistry registry) {
+        // 消息队列Queue绑定
+        for (ChatQueueType chatQueueType : ChatQueueType.values()) {
+            Queue queue = new Queue(chatQueueType.getQueueName(), true);
+            RootBeanDefinition queueDef = new RootBeanDefinition();
+            queueDef.setInstanceSupplier(() -> queue);
+            queueDef.setBeanClass(Queue.class);
+            registry.registerBeanDefinition(queue.getName(), queueDef);
+        }
+    }
+
+    @Bean
+    public MessageConverter rabbitMessageConverter() {
+        return new ChatMQMessageConverter();
+    }
+
+    @Bean
+    public RabbitTemplate rabbitTemplate(CachingConnectionFactory connectionFactory,
+                                         MessageConverter messageConverter) {
+        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
+        rabbitTemplate.setMessageConverter(messageConverter);
+        return rabbitTemplate;
+    }
+
+}

im-das-api/im-das-user-api/src/main/java/net/sopod/soim/das/user/api/mq/ChatMQ.java

@@ -1,15 +0,0 @@
-package net.sopod.soim.das.user.api.mq;
-
-/**
- * ChatMQ
- *
- * @author tmy
- * @date 2022-05-30 17:43
- */
-public class ChatMQ {
-
-    private MQType mqType;
-
-    private byte[] body;
-
-}

im-das-api/im-das-user-api/src/main/java/net/sopod/soim/das/user/api/mq/ChatMQMessageConverter.java

@@ -0,0 +1,76 @@
+package net.sopod.soim.das.user.api.mq;
+
+import net.sopod.soim.common.constant.Consts;
+import net.sopod.soim.common.util.Converter;
+import net.sopod.soim.common.util.ImClock;
+import net.sopod.soim.common.util.Jackson;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.support.converter.AbstractMessageConverter;
+import org.springframework.amqp.support.converter.MessageConversionException;
+import org.xerial.snappy.Snappy;
+
+import java.io.IOException;
+
+/**
+ * ChatMessageConverter
+ * 聊天消息到 MQ 生产和消费序列化 {@link ChatQueueType}
+ *
+ * @author tmy
+ * @date 2022-05-30 18:07
+ */
+public class ChatMQMessageConverter extends AbstractMessageConverter {
+
+    private static final Logger logger = LoggerFactory.getLogger(ChatMQMessageConverter.class);
+
+    public static final String MQ_MSG_TYPE = "MQ_MSG_TYPE";
+
+    public static final String MQ_MSG_IS_SNAPPY = "MQ_MSG_IS_SNAPPY";
+
+    @Override
+    protected Message createMessage(Object msg, MessageProperties messageProperties) {
+        if (msg instanceof Message) {
+            return (Message) msg;
+        }
+        ChatQueueType chatQueueType = ChatQueueType.getMQType(msg.getClass(),
+                type -> new MessageConversionException(String.format("不支持%s消息类型,在MQType添加", type.getName())));
+        messageProperties.setHeader(MQ_MSG_TYPE, chatQueueType.ordinal());
+        messageProperties.setMessageId(chatQueueType.getMsgId(msg));
+        // 消息序列化成字节
+        byte[] bytes = Jackson.msgpack().serializeBytes(msg);
+        if (bytes.length > Consts.KB) {
+            // snappy 压缩
+            try {
+                long start = ImClock.millis();
+                byte[] compressed = Snappy.compress(bytes);
+                logger.info("snappy compress msg bytes {} to {}, use {} ms.", bytes.length, compressed.length, ImClock.millis() - start);
+                bytes = compressed;
+                messageProperties.setHeader(MQ_MSG_IS_SNAPPY, true);
+            } catch (IOException e) {
+                logger.error("snappy compress error: bytes len {}", bytes.length, e);
+            }
+        }
+        return new Message(bytes, messageProperties);
+    }
+
+    @Override
+    public Object fromMessage(Message message) throws MessageConversionException {
+        MessageProperties messageProperties = message.getMessageProperties();
+        Object ordinal = messageProperties.getHeaders().get(MQ_MSG_TYPE);
+        Object isSnappy = messageProperties.getHeaders().get(MQ_MSG_IS_SNAPPY);
+        ChatQueueType chatQueueType = ChatQueueType.getMQType(Converter.toInt(ordinal),
+                ori -> new MessageConversionException(String.format("没有ordinal为%d的消息类型,在MQType添加", ori)));
+        byte[] body = message.getBody();
+        if (Boolean.TRUE.equals(isSnappy)) {
+            try {
+                body = Snappy.uncompress(body);
+            } catch (IOException e) {
+                throw new MessageConversionException(String.format("消息设置为snappy压缩但解压失败,消息长度 %d bytes", body.length), e);
+            }
+        }
+        return Jackson.msgpack().deserializeBytes(body, chatQueueType.getMsgType());
+    }
+
+}

im-das-api/im-das-user-api/src/main/java/net/sopod/soim/das/user/api/mq/ChatMessageConverter.java

@@ -1,39 +0,0 @@
-package net.sopod.soim.das.user.api.mq;
-
-import net.sopod.soim.common.constant.Consts;
-import net.sopod.soim.common.util.Jackson;
-import net.sopod.soim.das.user.api.model.entity.ImMessage;
-import org.springframework.amqp.core.Message;
-import org.springframework.amqp.core.MessageProperties;
-import org.springframework.amqp.support.converter.AbstractMessageConverter;
-import org.springframework.amqp.support.converter.MessageConversionException;
-
-/**
- * ChatMessageConverter
- *
- * @author tmy
- * @date 2022-05-30 18:07
- */
-public class ChatMessageConverter extends AbstractMessageConverter {
-
-    @Override
-    protected Message createMessage(Object object, MessageProperties messageProperties) {
-        if (object instanceof Message) {
-            return (Message) object;
-        }
-        byte[] bytes = Jackson.msgpack().serializeBytes(object);
-        if (bytes.length > 100 * Consts.KB) {
-            // TODO 压缩
-        }
-
-        return null;
-    }
-
-    @Override
-    public Object fromMessage(Message message) throws MessageConversionException {
-        byte[] body = message.getBody();
-        Jackson.msgpack().deserializeBytes(body, ImMessage.class);
-        return null;
-    }
-
-}

im-das-api/im-das-user-api/src/main/java/net/sopod/soim/das/user/api/mq/ChatMsg.java

@@ -1,27 +0,0 @@
-package net.sopod.soim.das.user.api.mq;
-
-import net.sopod.soim.common.util.Jackson;
-import net.sopod.soim.common.util.StringUtil;
-import net.sopod.soim.das.user.api.model.entity.ImMessage;
-import org.springframework.amqp.core.Message;
-import org.springframework.amqp.core.MessageProperties;
-
-/**
- * ChatMsg
- *
- * @author tmy
- * @date 2022-05-30 15:03
- */
-public class ChatMsg extends Message {
-
-    public ChatMsg(ImMessage imMessage) {
-        super(Jackson.msgpack().serializeBytes(imMessage), getMessageProperties(imMessage));
-    }
-
-    private static MessageProperties getMessageProperties(ImMessage imMessage) {
-        MessageProperties messageProperties = new MessageProperties();
-        messageProperties.setMessageId(StringUtil.toString(imMessage.getId()));
-        return messageProperties;
-    }
-
-}

im-das-api/im-das-user-api/src/main/java/net/sopod/soim/das/user/api/mq/ChatQueue.java

@@ -0,0 +1,14 @@
+package net.sopod.soim.das.user.api.mq;
+
+/**
+ * ChatQueue
+ *
+ * @author tmy
+ * @date 2022-05-31 00:05
+ */
+public class ChatQueue {
+
+    public static final String IM_MESSAGE_PERSISTENT = "IM_MESSAGE_PERSISTENT";
+
+    public static final String IM_GROUP_MESSAGE_PERSISTENT = "IM_GROUP_MESSAGE_PERSISTENT";
+}

im-das-api/im-das-user-api/src/main/java/net/sopod/soim/das/user/api/mq/ChatQueueType.java

@@ -0,0 +1,73 @@
+package net.sopod.soim.das.user.api.mq;
+
+import net.sopod.soim.common.util.Converter;
+import net.sopod.soim.common.util.StringUtil;
+import net.sopod.soim.das.user.api.model.entity.ImGroupMessage;
+import net.sopod.soim.das.user.api.model.entity.ImMessage;
+
+import java.util.function.Function;
+
+/**
+ * MQType
+ *
+ * @author tmy
+ * @date 2022-05-30 17:43
+ */
+public enum ChatQueueType {
+
+    /** 单聊消息 */
+    IM_MESSAGE(ChatQueue.IM_MESSAGE_PERSISTENT,
+            ImMessage.class,
+            msg -> Converter.toString(((ImMessage)msg).getId())),
+
+    /** 群聊消息 */
+    IM_GROUP_MESSAGE(ChatQueue.IM_GROUP_MESSAGE_PERSISTENT,
+            ImGroupMessage.class,
+            msg -> Converter.toString(((ImGroupMessage)msg).getId()));
+
+    private final Class<?> msgType;
+
+    private final String queueName;
+
+    private final Function<Object, String> msgIdConverter;
+
+    public static final String TOPIC_CHAT_PERSISTENT = "topic.chat_persistent";
+
+    ChatQueueType(String queueName, Class<?> msgType, Function<Object, String> msgIdConverter) {
+        this.queueName = queueName;
+        this.msgType = msgType;
+        this.msgIdConverter = msgIdConverter;
+    }
+
+    public Class<?> getMsgType() {
+        return msgType;
+    }
+
+    public String getQueueName() {
+        return queueName;
+    }
+
+    public String getMsgId(Object msg) {
+        String msgId = msgIdConverter.apply(msg);
+        return msgId != null ? msgId : StringUtil.randomUUID();
+    }
+
+    public static ChatQueueType getMQType(Class<?> msgType, Function<Class<?>, RuntimeException> exceptionProvider) {
+        ChatQueueType[] values = values();
+        for (ChatQueueType type : values) {
+            if (type.msgType == msgType) {
+                return type;
+            }
+        }
+        throw exceptionProvider.apply(msgType);
+    }
+
+    public static ChatQueueType getMQType(int ordinal, Function<Integer, RuntimeException> exceptionProvider) {
+        ChatQueueType[] values = values();
+        if (ordinal < values.length) {
+            return values[ordinal];
+        }
+        throw exceptionProvider.apply(ordinal);
+    }
+
+}

im-das-api/im-das-user-api/src/main/java/net/sopod/soim/das/user/api/mq/MQType.java

@@ -1,47 +0,0 @@
-package net.sopod.soim.das.user.api.mq;
-
-import net.sopod.soim.das.user.api.model.entity.ImGroupMessage;
-import net.sopod.soim.das.user.api.model.entity.ImMessage;
-
-/**
- * MQType
- *
- * @author tmy
- * @date 2022-05-30 17:43
- */
-public enum MQType {
-
-    /** 单聊消息 */
-    IM_MESSAGE("IM_MESSAGE", ImMessage.class),
-
-    /** 群聊消息 */
-    IM_GROUP_MESSAGE("IM_GROUP_MESSAGE", ImGroupMessage.class);
-
-    private final Class<?> beanClazz;
-
-    private final String queueName;
-
-    MQType(String queueName, Class<?> beanClazz) {
-        this.queueName = queueName;
-        this.beanClazz = beanClazz;
-    }
-
-    public Class<?> getBeanClazz() {
-        return beanClazz;
-    }
-
-    public String getQueueName() {
-        return queueName;
-    }
-
-    public static MQType getMQType(Class<?> beanClazz) {
-        MQType[] values = values();
-        for (MQType type : values) {
-            if (type.beanClazz == beanClazz) {
-                return type;
-            }
-        }
-        return null;
-    }
-
-}

im-das-api/im-das-user-api/src/main/java/net/sopod/soim/das/user/api/mq/package-info.java

@@ -0,0 +1,3 @@
+
+@org.springframework.lang.NonNullApi
+package net.sopod.soim.das.user.api.mq;

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

@@ -0,0 +1 @@
+org.springframework.boot.autoconfigure.EnableAutoConfiguration=net.sopod.soim.logic.api.segmentid.config.SegmentIdAutoConfiguration

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

@@ -1,7 +1,9 @@
 package net.sopod.soim.das.user.amqp;
 
 import net.sopod.soim.common.util.Jackson;
+import net.sopod.soim.das.user.api.model.entity.ImGroupMessage;
 import net.sopod.soim.das.user.api.model.entity.ImMessage;
+import net.sopod.soim.das.user.api.mq.ChatQueue;
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.rabbit.annotation.RabbitHandler;
 import org.springframework.amqp.rabbit.annotation.RabbitListener;
@@ -14,21 +16,18 @@ import org.springframework.stereotype.Component;
  * @author tmy
  * @date 2022-05-28 16:26
  */
+@RabbitListener(queues = {ChatQueue.IM_MESSAGE_PERSISTENT, ChatQueue.IM_GROUP_MESSAGE_PERSISTENT})
 @Component
-@RabbitListener(queues = "hello")
 public class Receiver {
 
     @RabbitHandler
-    public void process(Message message) {
+    public void process(ImGroupMessage imGroupMessage) {
-        byte[] body = message.getBody();
+        System.out.println("receiver2: " + imGroupMessage);
-        ImMessage imMessage = Jackson.msgpack().deserializeBytes(body, ImMessage.class);
-        System.out.println("Receiver: " + imMessage);
     }
 
     @RabbitHandler
-    public void process(byte[] payload) {
+    public void process(ImMessage imMessage) {
-        ImMessage imMessage = Jackson.msgpack().deserializeBytes(payload, ImMessage.class);
+        System.out.println("receive3: " + imMessage);
-        System.out.println("receive2: " + imMessage);
     }
 
 }

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

@@ -1,14 +1,18 @@
 package net.sopod.soim.das.user.amqp;
 
 import net.sopod.soim.common.util.ImClock;
+import net.sopod.soim.das.user.api.model.entity.ImGroupMessage;
 import net.sopod.soim.das.user.api.model.entity.ImMessage;
-import net.sopod.soim.das.user.api.mq.ChatMsg;
+import net.sopod.soim.das.user.api.mq.ChatQueueType;
 import org.springframework.amqp.core.AmqpTemplate;
 import org.springframework.boot.context.event.ApplicationReadyEvent;
 import org.springframework.context.ApplicationListener;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.stereotype.Component;
 
+import java.util.Random;
+import java.util.concurrent.TimeUnit;
+
 /**
  * Sender
  *
@@ -22,18 +26,34 @@ public class Sender implements ApplicationListener<ApplicationReadyEvent> {
     public void onApplicationEvent(ApplicationReadyEvent event) {
         ConfigurableApplicationContext context = event.getApplicationContext();
         AmqpTemplate amqpTemplate = context.getBean(AmqpTemplate.class);
-        // amqpTemplate.convertAndSend("hello", "hello:" + ImClock.millis());
+        for (int i = 0; i < 10; i++) {
-        ImMessage imMessage = new ImMessage()
+            try {
-                .setId(10086L)
+                TimeUnit.MILLISECONDS.sleep(500);
-                .setSender(10808L)
+            } catch (InterruptedException e) {
-                .setReceiver(180002L)
+                e.printStackTrace();
-                .setFriendId(101010L)
+            }
-                .setContent("何时")
+            System.out.println("send msg....");
-                .setCreateTime(ImClock.date());
+            String content = "《蒙娜丽莎》的姿势，已经升华为一种神圣的符号，它不仅仅是艺术规律问题，不仅仅是对文艺复兴人文思想粗糙大潮的冲击，而是心灵回归者、觉醒者心中的一个崇高细致理想的显现。《蒙娜丽莎》具有另一种难以说明的象征意义，如佛的坐势、站势、讲道姿势，菩萨的动势，基督的几种象征性姿势：如两手摊开，一手持十字，一手指天，这有某种非世俗的意味，不是做作，而是不得不是如此，是最不自然最自然，最不简单最简单，最轻松最沉甸甸的样式，你无法赞美它，又无法挑剔它，它就是这样，原始的样子，人们只能沉醉其中，而不能有条理地褒贬它。一切欺骗与神秘都集中在《蒙娜丽莎》身上，而令人们产生一种畏惧。《蒙娜丽莎》是一个精力充沛的形象，在该幅作品面前，不能说她是女人肖像，她穿越一切无所不见、又视而不见的空洞目光，想躲避是办不到的，而想迎接那目光同样也是枉然。奇异的前额，广阔得失去了一个平凡人的味道，弥漫在脸上的那种神奇表情，似笑、非笑使人们无法相信这是一张现实的脸，而它的存在，又无法使人们的怀疑进行得彻底。《蒙娜丽莎》的表情，像东方佛教中佛陀的表情一样，是非凡人所能做出来的。“蒙娜丽莎”不是因为了什么而微笑，她只是静静地在那，脸上是自然地出现的一种永恒的、无所谓表情的表情。 [12]  《蒙娜丽莎》的美学意义，主要在于人物形象焕发出的人性的光辉；而在这之前，即使是人的形象，也带有或多或少的宗教气息。 [13] ";
-        ChatMsg chatMsg = new ChatMsg(imMessage);
+            if (i % 2 == 0) {
-        amqpTemplate.convertAndSend("hello", 123);
+                ImGroupMessage imGroupMessage = new ImGroupMessage()
-        amqpTemplate.send("hello", chatMsg);
+                        .setId(new Random().nextLong())
-        System.out.println("send msg....");
+                        .setGroupId(10010L)
+                        .setSender(10086L)
+                        .setCreateTime(ImClock.date())
+                        .setContent(content);
+                amqpTemplate.convertAndSend(ChatQueueType.IM_GROUP_MESSAGE.getQueueName(), imGroupMessage);
+                continue;
+            }
+            ImMessage imMessage = new ImMessage()
+                    .setId(new Random().nextLong())
+                    .setSender(10808L)
+                    .setReceiver(180002L)
+                    .setFriendId(101010L)
+                    .setContent(content)
+                    .setCreateTime(ImClock.date());
+            amqpTemplate.convertAndSend(ChatQueueType.IM_MESSAGE.getQueueName(), imMessage);
+
+        }
     }
 
 }

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

@@ -1,30 +0,0 @@
-package net.sopod.soim.das.user.config;
-
-import org.springframework.amqp.core.Queue;
-import org.springframework.amqp.rabbit.core.RabbitTemplate;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-
-/**
- * RabbitConfig
- * RabbitMQ的配置类，用来配置队列、交换器、路由等高级信息
- *
- * @author tmy
- * @date 2022-05-28 16:31
- */
-@Configuration
-public class RabbitConfig {
-
-    @Bean
-    public Queue helloQueue() {
-        return new Queue("hello");
-    }
-
-    @Bean
-    public RabbitTemplate rabbitTemplate() {
-        RabbitTemplate rabbitTemplate = new RabbitTemplate();
-        rabbitTemplate.setMessageConverter(null);
-        return null;
-    }
-
-}