You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
116 lines
3.7 KiB
116 lines
3.7 KiB
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;
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
}
|
|
|