redis 消费

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

+package com.liquidnet.service.consumer.sweet.config;
+
+import com.liquidnet.service.consumer.sweet.receiver.ConsumerCityVoteRdsReceiver;
+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_CITY_VOTE_DRAW;
+
+@Configuration
+public class ConsumerCityVoteRedisStreamConfig {
+    @Autowired
+    ConsumerCityVoteRdsReceiver consumerCityVoteRdsReceiver;
+
+    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 receiveSqlCityVote(StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer, int t) {
+        return listenerContainer.receiveAutoAck(
+                Consumer.from(SWEET_CITY_VOTE_DRAW.getGroup(), SWEET_CITY_VOTE_DRAW.name() + t),
+                StreamOffset.create(SWEET_CITY_VOTE_DRAW.getKey(), ReadOffset.lastConsumed()), consumerCityVoteRdsReceiver
+        );
+    }
+
+    /* —————————————————————————— ｜ —————————————————————————— ｜ —————————————————————————— */
+
+    /* -------------------------------------------------------- ｜ 缺票登记 */
+
+    @Bean
+    public Subscription subscriptionSqlCityVote(RedisConnectionFactory factory) {
+        var listenerContainer = this.buildStreamMessageListenerContainer(factory);
+        var subscription = receiveSqlCityVote(listenerContainer, 1);
+        listenerContainer.start();
+        return subscription;
+    }
+
+    @Bean
+    public Subscription subscriptionSqlCityVote2(RedisConnectionFactory factory) {
+        var listenerContainer = this.buildStreamMessageListenerContainer(factory);
+        var subscription = receiveSqlCityVote(listenerContainer, 1);
+        listenerContainer.start();
+        return subscription;
+    }
+
+    @Bean
+    public Subscription subscriptionSqlCityVote3(RedisConnectionFactory factory) {
+        var listenerContainer = this.buildStreamMessageListenerContainer(factory);
+        var subscription = receiveSqlCityVote(listenerContainer, 1);
+        listenerContainer.start();
+        return subscription;
+    }
+
+    /* -------------------------------------------------------- ｜  */
+}

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

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