WxPay/.svn/pristine/1d/1d27b6f46ffebe551ae8749c5bb570e347cfc872.svn-base

package com.ynxbd.common.helper.redis;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.StreamEntryID;
import redis.clients.jedis.params.XReadGroupParams;
import redis.clients.jedis.resps.StreamEntry;
import redis.clients.jedis.resps.StreamGroupInfo;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class RedisStreamHelper {

    public static void main(String[] args) {
        Jedis jedis = new Jedis();
    }

    private static JedisPool jedisPool;

    public static void initPool() {
//        JedisPoolConfig poolConfig = new JedisPoolConfig();
//        poolConfig.setMaxTotal(100);
//        poolConfig.setMaxIdle(10);
//        jedisPool = new JedisPool(poolConfig, host, port, timeout, password);
    }

    //将数据传入stream
    public StreamEntryID setStream(String key, Map<String, String> content) {
        try (Jedis jedis = jedisPool.getResource()) {
            StreamEntryID id = jedis.xadd("key", StreamEntryID.NEW_ENTRY, content);
            String resp = id.toString();
            System.out.println(resp);
            return id;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }


    // 创建消费者群组
    public void createConsumeGroup(String key, String groupName, String lastId) {
        try (Jedis jedis = jedisPool.getResource()) {
            StreamEntryID id;
            if (lastId == null) {
                lastId = "0-0";
            }
            id = new StreamEntryID(lastId);
            // 没用消费者时是否自动创建，如果有，在创建会异常
            String resp = jedis.xgroupCreate(key, groupName, id, false);
            System.out.println(resp);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    // 消费者消费
    public StreamEntryID consume(String key, String groupName, String consumeName, String lastId) {
        try (Jedis jedis = jedisPool.getResource()) {
            /*
              COUNT 最多读取多少条消息
              BLOCK 是否已阻塞的方式读取消息，默认不阻塞，如果milliseconds设置为0，表示永远阻塞
             */
            XReadGroupParams xReadGroupParams = new XReadGroupParams().block(0).count(1);
            Map<String, StreamEntryID> streams = new HashMap<>();
            streams.put(key, StreamEntryID.UNRECEIVED_ENTRY); // >
            List<Map.Entry<String, List<StreamEntry>>>
                    entries = jedis.xreadGroup(groupName, consumeName, xReadGroupParams, streams);

            System.out.println(entries);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }



    // 消息确认
    public void ackMsg(String key, String groupName,StreamEntryID streamEntryID) {
        if (streamEntryID == null) {
            return;
        }
        try (Jedis jedis = jedisPool.getResource()) {
            jedis.xack(key, groupName, streamEntryID);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }



    // 检查消费者组是否存在
    public boolean checkGroup(String key, String groupName) {
        try (Jedis jedis = jedisPool.getResource()) {
            List<StreamGroupInfo> groupInfos = jedis.xinfoGroups(key);
            for (StreamGroupInfo item : groupInfos) {
                if (item.getName().equals(groupName)) {
                    return true;
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }






}