WPF下使用FreeRedis操作RedisStream实现简单的消息队列

Redis Stream简介

Redis Stream是随着5.0版本发布的一种新的Redis数据类型：

高效消费者组：允许多个消费者组从同一数据流的不同部分消费数据，每个消费者组都能独立地处理消息，这样可以并行处理和提高效率。

阻塞操作：消费者可以设置阻塞操作，这样它们会在流中有新数据添加时被唤醒并开始处理，这有助于减少资源消耗并提高响应速度。

数据持久化：它可以将数据持久化到内存（配置本地持久化后会写入到存储设备）中进行保存，等待消费。

多生产者多消费者：Redis Streams能够在多个生产者和消费者之间建立一个数据通道，使得数据的流动和处理更加灵活。

扩展性和异步通信：用户可以通过应用程序轻松扩展消费者数量，并且生产者和消费者之间的通信可以是异步的，这有助于提高系统的整体性能。

满足多样化需求：Redis Streams满足从实时数据处理到历史数据访问的各种需求，同时保持易于管理。

Redis Stream可以干什么

消息队列：Redis Stream可以用作一个可靠的消息队列系统，支持发布/订阅模式，生产者和消费者可以异步地发送和接收消息。

任务调度：Redis Stream可以用于实现分布式任务调度系统，将任务分发到多个消费者进行处理，从而提高处理速度和系统可扩展性。

事件驱动架构：Redis Stream可以作为事件驱动架构中的核心组件，用于处理来自不同服务的事件，实现解耦和灵活性。

FreeRedis简介

FreeRedis 的命名来自，“自由”、“免费”，它和名字与 FreeSql 是一个理念，简易是他们一致的追寻方向，最低可支持 .NET Framework 4.0 运行环境，支持到 Redis-server 7.2。

github MIT开源协议

作者博客园地址

官方介绍

基于 .NET 的 Redis 客户端，支持 .NET Core 2.1+、.NET Framework 4.0+、Xamarin 以及 AOT。

🌈 所有方法名与 redis-cli 保持一致
🌌 支持 Redis 集群（服务端要求 3.2 及以上版本）
⛳ 支持 Redis 哨兵模式
🎣 支持主从分离（Master-Slave）
📡 支持发布订阅（Pub-Sub）
📃 支持 Redis Lua 脚本
💻 支持管道（Pipeline）
📰 支持事务
🌴 支持 GEO 命令（服务端要求 3.2 及以上版本）
🌲 支持 STREAM 类型命令（服务端要求 5.0 及以上版本）
⚡ 支持本地缓存（Client-side-cahing，服务端要求 6.0 及以上版本）
🌳 支持 Redis 6 的 RESP3 协议

要实现的功能

1、生产者生产数据

2、消费者消费数据后确认

3、消费者消费数据后不确认

4、已消费但超时未确认的消息监控

5、已消费但超时未确认的消息二次消费

项目依赖

WPF
CommunityToolkit.Mvvm
FreeRedis
Newtonsoft.Json
NLog
redis-windows-7.2.5

业务场景代码

涉及到的Redis命令

创建消费者组 XGROUP CREATE key group [MKSTREAM] [ENTRIESREAD entries-read]

查询消费者组信息 XINFO STREAM key [FULL [COUNT count]]

消息队列数量（长度） XLEN key

添加消息到队列尾部 XADD key [NOMKSTREAM] [ [= | ~] threshold [LIMIT count]] field value [field value ...]

消费组成员读取消息 XREADGROUP GROUP group consumer [COUNT count] [BLOCK milliseconds] [NOACK] STREAMS key [key ...] id [id ...]

确认消息 XACK key group id [id ...]

删除消息 XDEL key id [id ...]

获取消费未确认消息的队列信息 XPENDING key group [[IDLE min-idle-time] start end count [consumer]]

把未消费的消息消费者转移到当前消费者名下 XAUTOCLAIM key group consumer min-idle-time start [COUNT count] [JUSTID]

代码

App.xaml.cs

    public partial class App : Application
    {
        public static Logger Logger = LogManager.GetCurrentClassLogger();        
        public static RedisClient RedisHelper;
        public static MainViewModel MainViewModel;
        private void App_OnStartup(object sender, StartupEventArgs e)
        {
            Current.DispatcherUnhandledException += Current_DispatcherUnhandledException;

            try
            {
                //redis6以上版本启用了ACL用户管理机制，默认用户名是default，可以忽略密码
                RedisHelper = new RedisClient("127.0.0.1:6379,user=defualt,defaultDatabase=13");
                RedisHelper.Serialize = JsonConvert.SerializeObject;//序列化
                RedisHelper.Deserialize = JsonConvert.DeserializeObject;//反序列化
                RedisHelper.Notice += (s, ee) => Console.WriteLine(ee.Log); //打印命令日志
                MainViewModel = new MainViewModel();
            }
            catch (Exception exception)
            {
                MessageBox.Show(exception.Message);
                Current.Shutdown(-100);
            }
        }

        private void Current_DispatcherUnhandledException(object sender, System.Windows.Threading.DispatcherUnhandledExceptionEventArgs e)
        {
            e.Handled = true;
            Logger.Error($"未捕获的错误：来源:{sender},错误：{e}");
        }

        private void App_OnExit(object sender, ExitEventArgs e)
        {
            RedisHelper.Dispose();
        }
    }

MainWindow.xaml的主要内容

MainWindow.xaml.cs

	public partial class MainWindow : Window
    {
        public MainWindow()
        {
            InitializeComponent();
            DataContext = App.MainViewModel;
        }

        private void MainWindow_OnLoaded(object sender, RoutedEventArgs e)
        {
            App.MainViewModel.RecordCount = 1000;
            App.MainViewModel.TaskCount = 5;
            App.MainViewModel.ConsumerCount = 1;
            App.MainViewModel.ConsumeInfo = "等待消费信息……";
            App.MainViewModel.PendingInfo = "加载中……";
            App.MainViewModel.StreamInfo = "加载中……";
        }
    }

MainViewModel的声明和变量定义

	public class MainViewModel : ObservableObject
	{
		#region 变量定义
		private readonly string _streamKey = "redisstream";
		private readonly string _consumeGroupName = "counsumeGroup";

		private DateTime _utcTime = new DateTime(1970, 1, 1, 0, 0, 0);
		/// 
		/// 生成的消息条数
		/// 
		private static int _exchangeValue;
		/// 
		/// 剩余未消费条数
		/// 
		private static int _consumeValue;
		/// 
		/// 消费信息展示队列
		/// 
		private static ConcurrentQueue _consumedQueue = new ConcurrentQueue();
		/// 
		/// 消费未确认展示队列
		/// 
		private static ConcurrentQueue _pendingConsumedQueue = new ConcurrentQueue();
		/// 
		/// 退出令牌
		/// 
		private CancellationTokenSource _cancellationTokenSource;
		/// 
		/// 生成消息
		/// 
		public RelayCommand ProducerRelayCommand { get; }
		/// 
		/// 消费消息
		/// 
		public RelayCommand ConsumeRelayCommand { get; }
		/// 
		/// 消费未确认信息队列消费
		/// 
		public RelayCommand PendingRelayCommand { get; }
		private int _recordCount;
		/// 
		/// 数据条数
		/// 
		public int RecordCount
		{
			get => _recordCount;
			set => SetProperty(ref _recordCount, value);
		}

		private int _taskCount;
		/// 
		/// 开启后台生产者数量
		/// 
		public int TaskCount
		{
			get => _taskCount;
			set => SetProperty(ref _taskCount, value);
		}

		private int _consumerCount;
		/// 
		/// 消费者数量
		/// 
		public int ConsumerCount
		{
			get => _consumerCount;
			set => SetProperty(ref _consumerCount, value);
		}

		private int _producerIndex;
		/// 
		/// 正在生产的序列号
		/// 
		public int ProducerIndex
		{
			get => Interlocked.Exchange(ref _producerIndex, _exchangeValue);
			set
			{
				SetProperty(ref _producerIndex, _exchangeValue);
			}
		}

		private long _consumeIndex;
		/// 
		/// 正在消费的序列号
		/// 
		public long ConsumeIndex
		{
			get => Interlocked.Read(ref _consumeIndex);
			set => SetProperty(ref _consumeIndex, value);
		}

		private string _streamInfo;
		/// 
		/// 队列信息展示
		/// 
		public string StreamInfo
		{
			get => _streamInfo;
			set => SetProperty(ref _streamInfo, value);
		}

		private string _consumeInfo;
		/// 
		/// 消费信息展示
		/// 
		public string ConsumeInfo
		{
			get => _consumeInfo;
			set
			{
				value = $"暂无消费消息[{DateTime.Now:yyyy-MM-dd HH:mm:ss}]";
				if (_consumedQueue.TryDequeue(out var message))
				{
					value = _consumeInfo + Environment.NewLine + message;
				}
				SetProperty(ref _consumeInfo, value);
			}
		}

		private string _pendingInfo;
		/// 
		/// 消费未确认队列信息展示
		/// 
		public string PendingInfo
		{
			get => _pendingInfo;
			set => SetProperty(ref _pendingInfo, value);
		}

		private string _pedingConsumeInfo;
		/// 
		/// 消费未确认队列的展示
		/// 
		public string PendingConsumeInfo
		{
			get => _pedingConsumeInfo;
			set
			{
				value = $"暂无未确认消费信息[{DateTime.Now:yyyy-MM-dd HH:mm:ss}]";
				if (_pendingConsumedQueue.TryDequeue(out var message))
				{
					value = _pedingConsumeInfo + Environment.NewLine + message;
				}

				SetProperty(ref _pedingConsumeInfo, value);
			}
		}

		private bool _isProduceCanExec;
		/// 
		/// 是否可以执行生成任务
		/// 
		public bool IsProduceCanExec
		{
			get => _isProduceCanExec;
			set
			{
				SetProperty(ref _isProduceCanExec, value);
				ProducerRelayCommand.NotifyCanExecuteChanged();
			}
		}

		private bool _isAutoAck;
		/// 
		/// 是否自动确认消费信息
		/// 
		public bool IsAutoAck
		{
			get => _isAutoAck;
			set
			{
				SetProperty(ref _isAutoAck, value);
				ProducerRelayCommand.NotifyCanExecuteChanged();
			}
		}
		#endregion

		public MainViewModel()
		{
			ProducerRelayCommand = new RelayCommand(async () => await DoProduce(), () => !_isProduceCanExec);

			ConsumeRelayCommand = new RelayCommand(async () => await DoConsume());

			PendingRelayCommand = new RelayCommand(async () => await DoPendingConsume());

			_cancellationTokenSource = new CancellationTokenSource();

			var exist = App.RedisHelper.Exists(_streamKey);
			if (!exist)
			{
				//创建消费组，同一个消费组可以有多个消费者，它们直接不会重复读取到同一条消息
				App.RedisHelper.XGroupCreate(_streamKey, _consumeGroupName, MkStream: true);
			}
			else
			{
				var groups = App.RedisHelper.XInfoGroups(_streamKey);

				if (groups == null || !groups.Any())
				{
					App.RedisHelper.XGroupCreate(_streamKey, _consumeGroupName);
				}
			}

			ConsumeIndex = App.RedisHelper.XLen(_streamKey);

			DoLoadStreamInfo();
			DoLoadPendingInfo();
		}
	}

消息生成者

private async Task DoProduce()
{
	if (IsProduceCanExec)
	{
		return;
	}
	IsProduceCanExec = true;

	if (TaskCount > RecordCount)
	{
		TaskCount = RecordCount;
	}

	_exchangeValue = 0;
	ProducerIndex = 0;

	if (TaskCount > 0)
	{
		var pageSize = RecordCount / TaskCount;
		var tasks = Enumerable.Range(1, TaskCount).Select(x =>
		{
			return Task.Run(() =>
			{
				var internalPageSize = pageSize;

				if (TaskCount > 1 && x == TaskCount)
				{
					if (x * internalPageSize  { { $"user_{x}", new MessageModel { Age = 16, Description = $"描述:{ProducerIndex}", Id = 1, Name = "wang", Status = 1 } } };
					App.RedisHelper.XAdd(_streamKey, 0, "*", dic);
				}
				return Task.CompletedTask;
			});
		});
		await Task.WhenAll(tasks);
	}
	IsProduceCanExec = false;
}

#endregion

消息消费者

private Task DoConsume()
{
	var groups = App.RedisHelper.XInfoGroups(_streamKey);
	if (groups == null || !groups.Any())
	{
		App.RedisHelper.XGroupCreate(_streamKey, _consumeGroupName);
	}

	//添加消费者
	var tasks = Enumerable.Range(1, _consumerCount).Select(c =>
	{
		var task = Task.Run(async () =>
		{
			//从消费组中读取消息，同一个组内的成员不会重复获取同一条消息。
			var streamRead = App.RedisHelper.XReadGroup(_consumeGroupName, $"consumer{c}", 0, _streamKey, ">");
			if (streamRead != null)
			{
				//取得消息
				var id = streamRead.id;
				var model = new Dictionary(1)
				{
					{ streamRead.fieldValues[0].ToString(), JsonConvert.DeserializeObject(streamRead.fieldValues[1].ToString()) }
				};
				_consumedQueue.Enqueue($"consumer{c}取到了消息{id},{DateTime.Now:yyyy-MM-dd HH:mm:ss}");
				ConsumeInfo = "";
				await Task.Delay(100);//模拟业务逻辑耗时
				
				if (IsAutoAck)
				{
					//ACK
					var success = App.RedisHelper.XAck(_streamKey, _consumeGroupName, id);
					if (success > 0)
					{
						//xdel
						App.RedisHelper.XDel(_streamKey, id);
						_consumedQueue.Enqueue($"consumer{c}成功消费了消息{id},{DateTime.Now:yyyy-MM-dd HH:mm:ss}");
					}
					else
					{
						_consumedQueue.Enqueue($"consumer{c}的消息{id}加入了未确认队列,{DateTime.Now:yyyy-MM-dd HH:mm:ss}");
					}
					ConsumeInfo = "";
				}
			}
		});
		return task;
	});
	Task.WhenAll(tasks);

	return Task.CompletedTask;
}
#endregion

未消费信息队列监控

private Task DoLoadStreamInfo()
{
	Task.Factory.StartNew(async () =>
	{
		while (!_cancellationTokenSource.IsCancellationRequested)
		{
			StreamInfo = "正在查询队列信息……";
			try
			{
				if (!App.RedisHelper.Exists(_streamKey))
				{
					StreamInfo = "队列尚未创建";
					await Task.Delay(3000);
					continue;
				}
				var info = App.RedisHelper.XInfoStream(_streamKey);
				StreamInfo = info?.first_entry == null ? "队列数据为空。" : $"队列长度：{info.length}{Environment.NewLine}第一个编号：{info.first_entry.id}{Environment.NewLine}最后一个编号：{info.last_entry.id}{Environment.NewLine}更新时间：{DateTime.Now:yyyy-MM-dd HH:mm:ss}";
				ConsumeIndex = info?.first_entry == null ? 0 : (int)info.length;
			}
			catch (Exception e)
			{
				StreamInfo = $"获取队列信息失败：{e.Message}";
			}
			await Task.Delay(3000);
		}
	}, _cancellationTokenSource.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default);
	return Task.CompletedTask;
}
#endregion

未消费成功（超时或业务逻辑执行失败）的消息队列消息展示

private Task DoLoadPendingInfo()
{
	Task.Run(async () =>
	{
		while (!_cancellationTokenSource.IsCancellationRequested)
		{
			PendingInfo = "正在查询未确认队列……";
			if (!App.RedisHelper.Exists(_streamKey))
			{
				PendingInfo = "队列尚未创建";
				await Task.Delay(3000);
				continue;
			}
			try
			{
				var info = App.RedisHelper.XPending(_streamKey, _consumeGroupName);
				if (info == null || info.count == 0)
				{
					PendingInfo = $"暂无未确认信息。[{DateTime.Now:yyyy-MM-dd HH:mm:ss}]";
					await Task.Delay(3000);

					continue;
				}
				var infoTxt = $"未消费数量：{info.count}{Environment.NewLine}涉及{info.consumers.Length}个消费者{Environment.NewLine}最小编号：{info.minId}{Environment.NewLine}最大编号：{info.maxId}{Environment.NewLine}更新时间：[{DateTime.Now:yyyy-MM-dd HH:mm:ss}]";
				PendingInfo = infoTxt;
			}
			catch (Exception e)
			{
				PendingInfo = $"获取队列信息失败：{e.Message}";
			}

			await Task.Delay(3000);
		}
	}, _cancellationTokenSource.Token);
	return Task.CompletedTask;
}
#endregion

未消费成功的消息重新消费

private Task DoPendingConsume()
{
	Task.Run(async () =>
	{
		while (!_cancellationTokenSource.IsCancellationRequested)
		{
			_pendingConsumedQueue.Enqueue($"正在查询未确认队列……[{DateTime.Now:yyyy-MM-dd HH:mm:ss}]");
			PendingConsumeInfo = "";
			if (!App.RedisHelper.Exists(_streamKey))
			{
				_pendingConsumedQueue.Enqueue($"队列尚未创建……[{DateTime.Now:yyyy-MM-dd HH:mm:ss}]");
				PendingConsumeInfo = "";
				await Task.Delay(3000);
				continue;
			}
			try
			{
				//从stream队列的头部（0-0的位置）获取2条已读取时间超过2分钟且未确认的消息，修改所有者为pendingUser重新消费并确认。
				var info = App.RedisHelper.XAutoClaim(_streamKey, _consumeGroupName, "pendingUser", 120000, "0-0", 2);
				if (info == null || info.entries == null || info.entries.Length == 0)
				{
					_pendingConsumedQueue.Enqueue("未确认队列中暂无信息。[{DateTime.Now:yyyy-MM-dd HH:mm:ss}]");
					await Task.Delay(3000);
					continue;
				}

				foreach (var entry in info.entries)
				{
					if (entry == null) continue;
					_pendingConsumedQueue.Enqueue($"未确认消费信息：{entry.id}[{DateTime.Now:yyyy-MM-dd HH:mm:ss}]");

					Debug.WriteLine(JsonConvert.DeserializeObject(entry.fieldValues[1].ToString()));
					PendingConsumeInfo = "";
					//ACK
					await Task.Delay(100);//模拟业务逻辑执行时间
					var success = App.RedisHelper.XAck(_streamKey, _consumeGroupName, entry.id);
					if (success > 0)
					{
						//xdel
						if (App.RedisHelper.XDel(_streamKey, entry.id) > 0)
						{
							_pendingConsumedQueue.Enqueue($"consumer[pendingUser]成功消费了消息{entry.id},{DateTime.Now:yyyy-MM-dd HH:mm:ss}");
						}
						else
						{
							_pendingConsumedQueue.Enqueue($"[pendingUser]删除{entry.id}[失败],{DateTime.Now:yyyy-MM-dd HH:mm:ss}");
						}
					}
					else
					{
						_pendingConsumedQueue.Enqueue($"[pendingUser]消费{entry.id}[失败],{DateTime.Now:yyyy-MM-dd HH:mm:ss}");
					}
				}
			}
			catch (Exception e)
			{
				PendingConsumeInfo = $"获取队列信息失败：{e.Message}";
			}
			PendingConsumeInfo = "";

			await Task.Delay(3000);
		}
	}, _cancellationTokenSource.Token);
	return Task.CompletedTask;
}

#endregion

总结

本次我们通过Redis的Stream数据类型实现了部署简单、高性能、高可用性的消息队列，在中小型项目上可适用于需要处理数据流转的场景。

参考资料

① Redis Streams

②Redis Commands

千百度

1.本站内容仅供参考，不作为任何法律依据。用户在使用本站内容时，应自行判断其真实性、准确性和完整性，并承担相应风险。

2.本站部分内容来源于互联网，仅用于交流学习研究知识，若侵犯了您的合法权益，请及时邮件或站内私信与本站联系，我们将尽快予以处理。

3.本文采用知识共享署名4.0国际许可协议 [BY-NC-SA] 进行授权

4.根据《计算机软件保护条例》第十七条规定“为了学习和研究软件内含的设计思想和原理，通过安装、显示、传输或者存储软件等方式使用软件的，可以不经软件著作权人许可，不向其支付报酬。”您需知晓本站所有内容资源均来源于网络，仅供用户交流学习与研究使用，版权归属原版权方所有，版权争议与本站无关，用户本人下载后不能用作商业或非法用途，需在24个小时之内从您的电脑中彻底删除上述内容，否则后果均由用户承担责任；如果您访问和下载此文件，表示您同意只将此文件用于参考、学习而非其他用途，否则一切后果请您自行承担，如果您喜欢该程序，请支持正版软件，购买注册，得到更好的正版服务。

5.本站是非经营性个人站点，所有软件信息均来自网络，所有资源仅供学习参考研究目的，并不贩卖软件，不存在任何商业目的及用途

THE END

NET开发学习