aspnet?core使用websocket實(shí)時(shí)更新商品信息的方法

更新時(shí)間：2024年01月06日 09:12:54 作者：星仔007

這篇文章主要介紹了aspnetcore使用websocket實(shí)時(shí)更新商品信息,本文通過實(shí)例代碼給大家介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值，需要的朋友參考下吧

先演示一下效果，再展示代碼邏輯。

中間幾次調(diào)用過程省略。。。

暫時(shí)只用到了下面四個(gè)項(xiàng)目

1.產(chǎn)品展示頁(yè)面中第一次通過接口去獲取數(shù)據(jù)庫(kù)的列表數(shù)據(jù)

/// <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.在前端頁(yè)面會(huì)把當(dāng)前頁(yè)面的產(chǎn)品列表id都發(fā)送到websocket中去

function updateAndSendProductIds(ids) {
    productIds = ids;
    // Check if the WebSocket is open
    if (socket.readyState === WebSocket.OPEN) {
        // Send the list of product IDs through the WebSocket connection
        socket.send(JSON.stringify(productIds));
    }
}
function fetchData() {
    const apiUrl = baseUrl + `/Catalog/items?pageSize=${pageSize}&pageIndex=${currentPage}`;
    axios.get(apiUrl)
        .then(response => {
            const data = response.data.data;
            displayProducts(baseUrl, data);
            const newProductIds = data.map(product => product.Id);
            // Check if the WebSocket is open
            updateAndSendProductIds(newProductIds);
            // 從響應(yīng)中獲取總頁(yè)數(shù)
            const totalPages = Math.ceil(response.data.count / pageSize);
            displayPagination(totalPages);
            // 更新當(dāng)前頁(yè)數(shù)的顯示
            const currentPageElement = document.getElementById('currentPage');
            currentPageElement.textContent = `當(dāng)前頁(yè)數(shù): ${currentPage + 1} / 總頁(yè)數(shù): ${totalPages}`;
        })
        .catch(error => {
            console.error('獲取數(shù)據(jù)失敗:', error);
        });
}

3.websocket拿到了id數(shù)據(jù)可以精確的把當(dāng)前頁(yè)面的產(chǎn)品都查出來再推送給product.html頁(yè)面，通過下面的ReceiveAsync方法獲取html發(fā)送的數(shù)據(jù)，再通過timer定時(shí)器每秒鐘Send方法實(shí)時(shí)的往頁(yè)面發(fā)送獲取到的數(shù)據(jù)，當(dāng)然這個(gè)是不斷的去從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>
    /// 實(shí)時(shí)推送產(chǎn)品主要是最新的庫(kù)存，其他信息也會(huì)更新
    /// </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)
        {
            // 獲取當(dāng)前時(shí)間并發(fā)送給所有連接的客戶端
            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))
                    {
                        //反序列化和構(gòu)造函數(shù)沖突，改造了一下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)
        {
            //每次頁(yè)面有刷新就會(huì)拿到展示的id列表
            productIdsStr = System.Text.Encoding.UTF8.GetString(buffer, 0, result.Count);
        }
    }
}

4.html頁(yè)面就可以拿到最新數(shù)據(jù)再去綁定到頁(yè)面

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);
});

整個(gè)流程就這么簡(jiǎn)單，但是這里需要保持?jǐn)?shù)據(jù)庫(kù)和redis的數(shù)據(jù)實(shí)時(shí)同步，否則頁(yè)面展示的就不是最新的數(shù)據(jù)就沒意義了。

再回到Catalog.Service服務(wù)中。

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

再做更新、新增、刪除等動(dòng)作的時(shí)候就調(diào)用一下DeleteCache方法，往后臺(tái)服務(wù)發(fā)送一個(gè)channel,當(dāng)后臺(tái)收到后就做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>
    /// 記得任何刪除了或者購(gòu)買了產(chǎn)品后需要?jiǎng)h除改產(chǎn)品的鍵
    /// </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.");
            }
        }
    }
}

這里還有優(yōu)化的空間可以只針對(duì)怕products的hashset的某個(gè)id去更新、刪除、新增一條數(shù)據(jù)。

示例代碼：

liuzhixin405/efcore-template (github.com)

到此這篇關(guān)于aspnetcore使用websocket實(shí)時(shí)更新商品信息的文章就介紹到這了,更多相關(guān)aspnetcore實(shí)時(shí)更新商品信息內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: