消息队列

liquidnet-bus-common/liquidnet-common-service-base/src/main/java/com/liquidnet/service/base/constant/MQConst.java

@@ -208,7 +208,30 @@ public class MQConst {
             return desc;
         }
     }
+    public enum SmileQueue {
+        SMILE_USER("smile:stream:sql.user", "group.smile.user", "用户的增删改");
+        private final String key;
+        private final String group;
+        private final String desc;
+
+        SmileQueue(String key, String group, String desc) {
+            this.key = key;
+            this.group = group;
+            this.desc = desc;
+        }
+
+        public String getKey() {
+            return key;
+        }
+
+        public String getGroup() {
+            return group;
+        }
 
+        public String getDesc() {
+            return desc;
+        }
+    }
     public enum ChimeQueue {
         USER_OPERATION_LIKE("chime:stream:rk.user.operation.like", "group.user.operation.like", "划卡操作-喜欢"),
         USER_OPERATION_DISLIKE("chime:stream:rk.user.operation.dislike", "group.user.operation.dislike", "划卡操作-不喜欢");

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-sweet/src/main/java/com/liquidnet/service/consumer/sweet/config/ConsumerSmileUserRedisStreamConfig.java

+package com.liquidnet.service.consumer.sweet.config;
+
+import com.liquidnet.service.base.constant.MQConst;
+import com.liquidnet.service.consumer.sweet.receiver.ConsumerSmileUserReceiver;
+import lombok.var;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.connection.stream.Consumer;
+import org.springframework.data.redis.connection.stream.MapRecord;
+import org.springframework.data.redis.connection.stream.ReadOffset;
+import org.springframework.data.redis.connection.stream.StreamOffset;
+import org.springframework.data.redis.stream.StreamMessageListenerContainer;
+import org.springframework.data.redis.stream.Subscription;
+
+import java.time.Duration;
+
+@Configuration
+public class ConsumerSmileUserRedisStreamConfig {
+    @Autowired
+    ConsumerSmileUserReceiver consumerSmileUserReceiver;
+
+    private StreamMessageListenerContainer<String, MapRecord<String, String, String>> buildStreamMessageListenerContainer(RedisConnectionFactory factory) {
+        var options = StreamMessageListenerContainer
+                .StreamMessageListenerContainerOptions
+                .builder()
+                .pollTimeout(Duration.ofMillis(1))
+                .build();
+        return StreamMessageListenerContainer.create(factory, options);
+    }
+
+    /**
+     * 用户的增加和修改
+     *
+     * @param listenerContainer
+     * @param t
+     * @return
+     */
+    private Subscription receiveSqlAnswer(StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer, int t) {
+        return listenerContainer.receiveAutoAck(
+                Consumer.from(MQConst.SmileQueue.SMILE_USER.getGroup(), MQConst.SmileQueue.SMILE_USER.name() + t),
+                StreamOffset.create(MQConst.SmileQueue.SMILE_USER.getKey(), ReadOffset.lastConsumed()), consumerSmileUserReceiver
+        );
+    }
+
+    /* —————————————————————————— ｜ —————————————————————————— ｜ —————————————————————————— */
+
+    /* -------------------------------------------------------- ｜ 缺票登记 */
+
+    @Bean
+    public Subscription subscriptionSqlAnswer(RedisConnectionFactory factory) {
+        var listenerContainer = this.buildStreamMessageListenerContainer(factory);
+        var subscription = receiveSqlAnswer(listenerContainer, 1);
+        listenerContainer.start();
+        return subscription;
+    }
+
+    @Bean
+    public Subscription subscriptionSqlAnswer2(RedisConnectionFactory factory) {
+        var listenerContainer = this.buildStreamMessageListenerContainer(factory);
+        var subscription = receiveSqlAnswer(listenerContainer, 2);
+        listenerContainer.start();
+        return subscription;
+    }
+
+    @Bean
+    public Subscription subscriptionSqlAnswer3(RedisConnectionFactory factory) {
+        var listenerContainer = this.buildStreamMessageListenerContainer(factory);
+        var subscription = receiveSqlAnswer(listenerContainer, 3);
+        listenerContainer.start();
+        return subscription;
+    }
+
+    /* -------------------------------------------------------- ｜  */
+}

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-sweet/src/main/java/com/liquidnet/service/consumer/sweet/receiver/ConsumerSmileUserReceiver.java

+package com.liquidnet.service.consumer.sweet.receiver;
+
+import com.liquidnet.service.base.constant.MQConst;
+import org.springframework.stereotype.Component;
+
+@Component
+public class ConsumerSmileUserReceiver extends AbstractSqlRedisReceiver{
+    @Override
+    protected String getRedisStreamKey() {
+        return MQConst.SmileQueue.SMILE_USER.getKey();
+    }
+
+    @Override
+    protected String getRedisStreamGroup() {
+        return MQConst.SmileQueue.SMILE_USER.getGroup();
+    }
+}

liquidnet-bus-service/liquidnet-service-smile/liquidnet-service-smile-impl/docu/redis_queue_create.txt

+XADD smile:stream:sql.user * 0 0
+XGROUP CREATE smile:stream:sql.user group.smile.user 0