aspnet core使用websocket实时更新商品信息的方法

先演示一下效果，再展示代码逻辑。

中间几次调用过程省略。。。

暂时只用到了下面四个项目

1.产品展示页面中第一次通过接口去获取数据库的列表数据

/// <summary>
/// 获取指定的商品目录
/// </summary>
/// <param name=\”pageSize\”></param>
/// <param name=\”pageIndex\”></param>
/// <param name=\”ids\”></param>
/// <returns></returns>
[HttpGet]
[Route(\”items\”)]
[ProducesResponseType(typeof(PaginatedViewModel<Catalog>), StatusCodes.Status200OK)]
[ProducesResponseType(typeof(IEnumerable<ProductDto>), StatusCodes.Status200OK)]
[ProducesResponseType(StatusCodes.Status400BadRequest)]
public async Task<IActionResult> Catalogs([FromQuery] int pageSize = 10, [FromQuery] int pageIndex = 0, string ids = null)
{
if (!string.IsNullOrEmpty(ids))
{
var items = await GetItemByIds(ids);
if (!items.Any())
{
return BadRequest(\”ids value invalid. Must be comma-separated list of numbers\”);
}
return Ok(items);
}
var totalItems = await _catalogContext.Catalogs
.LongCountAsync();
var itemsOnPage = await _catalogContext.Catalogs
.OrderBy(c => c.Name)
.Skip(pageSize * pageIndex)
.Take(pageSize)
.ToListAsync();
var result = itemsOnPage.Select(x => new ProductDto(x.Id.ToString(), x.Name, x.Price.ToString(), x.Stock.ToString(), x.ImgPath));
var model = new PaginatedViewModel<ProductDto>(pageIndex, pageSize, totalItems, result);
return Ok(model);
}

2.在前端页面会把当前页面的产品列表id都发送到websocket中去

3.websocket拿到了id数据可以精确的把当前页面的产品都查出来再推送给product.html页面，通过下面的ReceiveAsync方法获取html发送的数据，再通过timer定时器每秒钟Send方法实时的往页面发送获取到的数据，当然这个是不断的去从redis中去查的。

using System.Net.WebSockets;
using System.Threading.Tasks;
using System;
using WsServer.Handler;
using WsServer.Manager;
using StackExchange.Redis;
using Microsoft.Extensions.Configuration;
using System.Collections.Generic;
using Catalogs.Domain.Catalogs;
using Catalogs.Domain.Dtos;
using System.Net.Sockets;
namespace WebScoket.Server.Services
{
/// <summary>
/// 实时推送产品主要是最新的库存，其他信息也会更新
/// </summary>
public class ProductListHandler : WebSocketHandler
{
private System.Threading.Timer _timer;
private readonly IDatabase _redisDb;
//展示列表推送
private string productIdsStr;
public ProductListHandler(WebSocketConnectionManager webSocketConnectionManager,IConfiguration configuration) : base(webSocketConnectionManager)
{
ConnectionMultiplexer redis = ConnectionMultiplexer.Connect(configuration[\”DistributedRedis:ConnectionString\”] ?? throw new Exception(\”$未能获取distributedredis连接字符串\”));
_redisDb = redis.GetDatabase();
_timer = new System.Threading.Timer(Send, null, TimeSpan.Zero, TimeSpan.FromSeconds(1));
}
private void Send(object state)
{
// 获取当前时间并发送给所有连接的客户端
if (productIdsStr != null)
{
string[] productIds = System.Text.Json.JsonSerializer.Deserialize<string[]>(productIdsStr);
string hashKeyToRetrieve = \”products\”;
List<ProductDto> products = new List<ProductDto>();
foreach (var productId in productIds)
{
if(productId == \”null\”) {
continue;
}
string retrievedProductValue = _redisDb.HashGet(hashKeyToRetrieve, productId);
if (!string.IsNullOrEmpty(retrievedProductValue))
{
//反序列化和构造函数冲突，改造了一下Catalog
Catalog catalog = System.Text.Json.JsonSerializer.Deserialize<Catalog>(retrievedProductValue);
products.Add(new ProductDto(catalog.Id.ToString(), catalog.Name, catalog.Price.ToString(), catalog.Stock.ToString(), catalog.ImgPath));
}
}
if (products.Count > 0)
{
SendMessageToAllAsync(System.Text.Json.JsonSerializer.Serialize(products)).Wait();
}
else
{
SendMessageToAllAsync(\”NoProduct\”).Wait();
}
}
}
public override async Task ReceiveAsync(WebSocket socket, WebSocketReceiveResult result, byte[] buffer)
{
//每次页面有刷新就会拿到展示的id列表
productIdsStr = System.Text.Encoding.UTF8.GetString(buffer, 0, result.Count);
}
}
}

4.html页面就可以拿到最新数据再去绑定到页面

socket.addEventListener(\’message\’, (event) => {
if (event.data == \”NoProduct\”) {
clearProductList();
}
// Handle the received product data and update the product list
const productData = JSON.parse(event.data);
// Update the product list with the received data (call your displayProducts function)
displayProducts(baseUrl, productData);
});

整个流程就这么简单，但是这里需要保持数据库和redis的数据实时同步，否则页面展示的就不是最新的数据就没意义了。

再回到Catalog.Service服务中。

private async Task DeleteCache()
{
//await _redisDb.HashDeleteAsync(\”products\”,id); //没必要了
await _channel.Writer.WriteAsync(\”delete_catalog_fromredis\”);
}

再做更新、新增、删除等动作的时候就调用一下DeleteCache方法，往后台服务发送一个channel,当后台收到后就做redis删除并且从初始化sqlserver到redis列表同步的操作

using System.Reflection;
using System.Threading.Channels;
using Catalogs.Infrastructure.Database;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using StackExchange.Redis;
namespace Catalogs.WebApi.BackgroudServices
{
/// <summary>
/// 记得任何删除了或者购买了产品后需要删除改产品的键
/// </summary>
public class InitProductListToRedisService : BackgroundService
{
private readonly IServiceScopeFactory _serviceScopeFactory;
private readonly IDatabase _redisDb;
private readonly Channel<string> _channel;
private readonly ILogger _logger;
public InitProductListToRedisService(IServiceScopeFactory serviceScopeFactory, IConfiguration configuration, Channel<string> channel, ILogger<InitProductListToRedisService> logger)
{
_serviceScopeFactory = serviceScopeFactory;
ConnectionMultiplexer redis = ConnectionMultiplexer.Connect(configuration[\”DistributedRedis:ConnectionString\”] ?? throw new Exception(\”$未能获取distributedredis连接字符串\”));
_redisDb = redis.GetDatabase();
_channel = channel;
_logger = logger;
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
await Init();
while (!_channel.Reader.Completion.IsCompleted)
{
var msg = await _channel.Reader.ReadAsync();
if(msg == \”delete_catalog_fromredis\”)
{
await Init();
}
}
}
private async Task Init()
{
using var scope = _serviceScopeFactory.CreateScope();
try
{
CatalogContext _context = scope.ServiceProvider.GetRequiredService<CatalogContext>();
string hashKey = \”products\”;
var products = await _context.Catalogs.ToListAsync();
await _redisDb.KeyDeleteAsync(hashKey);
foreach (var product in products)
{
string productField = product.Id.ToString();
string productValue = System.Text.Json.JsonSerializer.Serialize(product);
_redisDb.HashSet(hashKey, new HashEntry[] { new HashEntry(productField, productValue) });
}
_logger.LogInformation($\”ProductList is over stored in Redis Hash.\”);
}
catch(Exception ex)
{
_logger.LogError($\”ProductLis stored in Redis Hash error.\”);
}
}
}
}