String boot 使用 redis stream 消息队列 - zhouted/zhouted.github.io GitHub Wiki

@Data
@AllArgsConstructor
@NoArgsConstructor
public class SampleObj {
    String name;
    ......
}

@Service
@Slf4j
public class SampleService {
    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    public RecordId produce(String streamKey, SampleObj obj ){
        ObjectRecord<String, SampleObj> record = StreamRecords.newRecord().ofObject(obj).withStreamKey(streamKey);
        RecordId recordId = this.redisTemplate.opsForStream().add(record);
        if (Objects.isNull(recordId)) {
            log.info("error sending {} event: {}", streamKey, obj);
            return null;
        }
        return recordId;
    }
}

@Configuration
@Slf4j
public class SampleRedisConfig {
    @Bean(initMethod = "start", destroyMethod = "stop")
    public StreamMessageListenerContainer<String, ObjectRecord<String, SampleObj>> streamMessageListenerContainer(
            RedisConnectionFactory connectionFactory
    ) {
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String,
                ObjectRecord<String, SampleObj>> options = StreamMessageListenerContainer
                .StreamMessageListenerContainerOptions
                .builder()
                .pollTimeout(Duration.ofMillis(100))
                .targetType(SampleObj.class)
                .build();

        StreamMessageListenerContainer<String, ObjectRecord<String, SampleObj>> container =
                StreamMessageListenerContainer.create(connectionFactory, options);
        return container;
    }
}

@Slf4j
@Service
public class AStreamListener implements StreamListener<String, ObjectRecord<String, SampleObj>>{
    private String streamKey = "stream_a";
    @Autowired
    private RedisTemplate<String, String> redisTemplate;
    @Autowired
    StreamMessageListenerContainer<String, ObjectRecord<String, SampleObj>> streamMessageListenerContainer;

    @PostConstruct
    public void init(){
        Subscription receive = streamMessageListenerContainer.receive(StreamOffset.create(streamKey, ReadOffset.latest()), this);
        log.info("register listener: {} {}", streamKey, receive);
    }

    @Override
    public void onMessage(ObjectRecord<String, SampleObj> message) {
        try {
            RecordId messageId = message.getId();
            SampleObj obj = message.getValue();
            log.info("收到消息：messageId={}, stream={}, body={}", messageId, streamKey, obj);
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            this.redisTemplate.opsForStream().delete(message);
        }
    }
}

private void createConsumerGroupIfNotExists(RedisConnectionFactory redisConnectionFactory,
                                            String streamKey, String groupName){
    try {
        try {
            redisConnectionFactory.getConnection().streamCommands()
                    .xGroupCreate(streamKey.getBytes(), groupName, ReadOffset.from("0-0"), true);
        } catch (RedisSystemException exception) {
            log.warn(exception.getCause().getMessage());
        }
    } catch (RedisSystemException ex){
        log.error(ex.getMessage());
    }
}

    @PostConstruct
    public void init(){
        createConsumerGroupIfNotExists(connectionFactory, streamKey, groupName);
        StreamOffset<String> streamOffset = StreamOffset.create(streamKey, ReadOffset.lastConsumed());
        String consumerName = InetAddress.getLocalHost().getHostName();
        Subscription receive = streamMessageListenerContainer.receive(Consumer.from(groupName, consumerName), streamOffset, this);
        log.info("register listener: {} {} {}", streamKey, groupName, receive);
    }

    @Override
    public void onMessage(ObjectRecord<String, SampleObj> message) {
        ......
        this.redisTemplate.opsForStream().acknowledge(streamKey, groupName, messageId.getValue());//delete(message);
        ......
    }