redis 消费

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

+
+--     使用     --
+XADD candy:stream:rk.coupon.use * 0 0
+XGROUP CREATE candy:stream:rk.coupon.use group.coupon.use 0
+
+--     回退     --
+XADD candy:stream:rk.coupon.back * 0 0
+XGROUP CREATE candy:stream:rk.coupon.back group.coupon.back 0
+
+--     领取     --
+XADD candy:stream:rk.coupon.receive * 0 0
+XGROUP CREATE candy:stream:rk.coupon.receive group.coupon.receive 0

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-kylin/src/main/java/com/liquidnet/service/consumer/kylin/config/ConsumerCandyCouponBackRedisStreamConfig.java

+package com.liquidnet.service.consumer.kylin.config;
+
+import com.liquidnet.service.consumer.kylin.receiver.ConsumerCandyCouponBackRdsReceiver;
+import com.liquidnet.service.consumer.kylin.receiver.ConsumerCandyCouponUseRdsReceiver;
+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;
+
+import static com.liquidnet.service.base.constant.MQConst.SweetQueue.SWEET_USER_INSERT_DRAW;
+
+@Configuration
+public class ConsumerCandyCouponBackRedisStreamConfig {
+    @Autowired
+    ConsumerCandyCouponBackRdsReceiver consumerCandyCouponBackRdsReceiver;
+
+    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 receiveSqlCandyCouponBack(StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer, int t) {
+        return listenerContainer.receiveAutoAck(
+                Consumer.from(SWEET_USER_INSERT_DRAW.getGroup(), SWEET_USER_INSERT_DRAW.name() + t),
+                StreamOffset.create(SWEET_USER_INSERT_DRAW.getKey(), ReadOffset.lastConsumed()), consumerCandyCouponBackRdsReceiver
+        );
+    }
+
+    /* —————————————————————————— ｜ —————————————————————————— ｜ —————————————————————————— */
+
+    /* -------------------------------------------------------- ｜ 缺票登记 */
+
+    @Bean
+    public Subscription subscriptionSqlCandyCouponBack(RedisConnectionFactory factory) {
+        var listenerContainer = this.buildStreamMessageListenerContainer(factory);
+        var subscription = receiveSqlCandyCouponBack(listenerContainer, 1);
+        listenerContainer.start();
+        return subscription;
+    }
+
+    @Bean
+    public Subscription subscriptionSqlCandyCouponBack2(RedisConnectionFactory factory) {
+        var listenerContainer = this.buildStreamMessageListenerContainer(factory);
+        var subscription = receiveSqlCandyCouponBack(listenerContainer, 1);
+        listenerContainer.start();
+        return subscription;
+    }
+
+    @Bean
+    public Subscription subscriptionSqlCandyCouponBack3(RedisConnectionFactory factory) {
+        var listenerContainer = this.buildStreamMessageListenerContainer(factory);
+        var subscription = receiveSqlCandyCouponBack(listenerContainer, 1);
+        listenerContainer.start();
+        return subscription;
+    }
+
+    /* -------------------------------------------------------- ｜  */
+}

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-kylin/src/main/java/com/liquidnet/service/consumer/kylin/config/ConsumerCandyCouponReceiveRedisStreamConfig.java

+package com.liquidnet.service.consumer.kylin.config;
+
+import com.liquidnet.service.consumer.kylin.receiver.ConsumerCandyCouponBackRdsReceiver;
+import com.liquidnet.service.consumer.kylin.receiver.ConsumerCandyCouponReceiveRdsReceiver;
+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;
+
+import static com.liquidnet.service.base.constant.MQConst.SweetQueue.SWEET_USER_INSERT_DRAW;
+
+@Configuration
+public class ConsumerCandyCouponReceiveRedisStreamConfig {
+    @Autowired
+    ConsumerCandyCouponReceiveRdsReceiver consumerCandyCouponReceiveRdsReceiver;
+
+    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 receiveSqlCandyCouponReceive(StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer, int t) {
+        return listenerContainer.receiveAutoAck(
+                Consumer.from(SWEET_USER_INSERT_DRAW.getGroup(), SWEET_USER_INSERT_DRAW.name() + t),
+                StreamOffset.create(SWEET_USER_INSERT_DRAW.getKey(), ReadOffset.lastConsumed()), consumerCandyCouponReceiveRdsReceiver
+        );
+    }
+
+    /* —————————————————————————— ｜ —————————————————————————— ｜ —————————————————————————— */
+
+    /* -------------------------------------------------------- ｜ 缺票登记 */
+
+    @Bean
+    public Subscription subscriptionSqlCandyCouponReceive(RedisConnectionFactory factory) {
+        var listenerContainer = this.buildStreamMessageListenerContainer(factory);
+        var subscription = receiveSqlCandyCouponReceive(listenerContainer, 1);
+        listenerContainer.start();
+        return subscription;
+    }
+
+    @Bean
+    public Subscription subscriptionSqlCandyCouponReceive2(RedisConnectionFactory factory) {
+        var listenerContainer = this.buildStreamMessageListenerContainer(factory);
+        var subscription = receiveSqlCandyCouponReceive(listenerContainer, 1);
+        listenerContainer.start();
+        return subscription;
+    }
+
+    @Bean
+    public Subscription subscriptionSqlCandyCouponBack3(RedisConnectionFactory factory) {
+        var listenerContainer = this.buildStreamMessageListenerContainer(factory);
+        var subscription = receiveSqlCandyCouponReceive(listenerContainer, 1);
+        listenerContainer.start();
+        return subscription;
+    }
+
+    /* -------------------------------------------------------- ｜  */
+}

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-kylin/src/main/java/com/liquidnet/service/consumer/kylin/config/ConsumerCandyCouponUseRedisStreamConfig.java

+package com.liquidnet.service.consumer.kylin.config;
+
+import com.liquidnet.service.consumer.kylin.receiver.ConsumerCandyCouponUseRdsReceiver;
+import com.liquidnet.service.consumer.kylin.receiver.ConsumerSweetUserRdsReceiver;
+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;
+
+import static com.liquidnet.service.base.constant.MQConst.SweetQueue.SWEET_USER_INSERT_DRAW;
+
+@Configuration
+public class ConsumerCandyCouponUseRedisStreamConfig {
+    @Autowired
+    ConsumerCandyCouponUseRdsReceiver consumerCandyCouponUseRdsReceiver;
+
+    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 receiveSqlCandyCouponUse(StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer, int t) {
+        return listenerContainer.receiveAutoAck(
+                Consumer.from(SWEET_USER_INSERT_DRAW.getGroup(), SWEET_USER_INSERT_DRAW.name() + t),
+                StreamOffset.create(SWEET_USER_INSERT_DRAW.getKey(), ReadOffset.lastConsumed()), consumerCandyCouponUseRdsReceiver
+        );
+    }
+
+    /* —————————————————————————— ｜ —————————————————————————— ｜ —————————————————————————— */
+
+    /* -------------------------------------------------------- ｜ 缺票登记 */
+
+    @Bean
+    public Subscription subscriptionSqlCandyCouponUse(RedisConnectionFactory factory) {
+        var listenerContainer = this.buildStreamMessageListenerContainer(factory);
+        var subscription = receiveSqlCandyCouponUse(listenerContainer, 1);
+        listenerContainer.start();
+        return subscription;
+    }
+
+    @Bean
+    public Subscription subscriptionSqlCandyCouponUse2(RedisConnectionFactory factory) {
+        var listenerContainer = this.buildStreamMessageListenerContainer(factory);
+        var subscription = receiveSqlCandyCouponUse(listenerContainer, 1);
+        listenerContainer.start();
+        return subscription;
+    }
+
+    @Bean
+    public Subscription subscriptionSqlCandyCouponUse3(RedisConnectionFactory factory) {
+        var listenerContainer = this.buildStreamMessageListenerContainer(factory);
+        var subscription = receiveSqlCandyCouponUse(listenerContainer, 1);
+        listenerContainer.start();
+        return subscription;
+    }
+
+    /* -------------------------------------------------------- ｜  */
+}

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-kylin/src/main/java/com/liquidnet/service/consumer/kylin/receiver/ConsumerCandyCouponBackRdsReceiver.java

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

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-kylin/src/main/java/com/liquidnet/service/consumer/kylin/receiver/ConsumerCandyCouponReceiveRdsReceiver.java

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

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-kylin/src/main/java/com/liquidnet/service/consumer/kylin/receiver/ConsumerCandyCouponUseRdsReceiver.java

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