组件适用

1.需要集群部署的服务，需要有主从的集群服务
2.如果是定时的任务，建议使用PasteTask任务调度器实现
3.组件适用于.net8.0以上的框架
4.集群对外的消息使用Channel队列发出

引入组件

1.先引入IHttpClientFactory,示例：context.Services.AddHttpClient();
2.添加对组件的应用 示例：<PackageReference Include="PasteCluster" Version="1.0.0" />
3.使用单例注入，示例：context.Services.AddSingleton<PasteClusterHandler>();
4.配置配置文件，示例：context.Services.Configure<PasteSloveConfig>(Configuration.GetSection("ClusterConfig"));

  "ClusterConfig": {
    "SloveToken": "zxcvfr43dr56hgt5",//集群密钥,可自定义，防止其他集群乱入！
    "ClusterHost": "",//已有的节点地址链路，示例http://192.110.0.3:80;http://192.110.0.7:80
    "CurrentHost": ""//当前节点的访问地址是多少 示例http://192.110.0.6:80
  }

以上是基础配置,需要自定义更多的请查阅PasteSloveConfig的属性

5.如果使用了Route路由，注意本组件需要/api/cluster/的路由转发配置

开始使用

1.初始化集群组件，写入必要的信息，比如集群中还有谁，自己的HOST是多少
2.定义一个HostedService用于接收集群产生的事件
示例代码:

using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using PasteCluster;
using TestCluster.redismodel;

namespace TestCluster
{
    /// <summary>
    /// 系统初始化
    /// </summary>
    public class StartHostedService : IHostedService
    {
        private readonly IServiceProvider _serviceProvider;
        private PasteClusterHandler _cluster_handler;
        private IAppCache _appCache;
        private ILogger<StartHostedService> _logger;
        /// <summary>
        /// 
        /// </summary>
        /// <param name="serviceProvider"></param>
        /// <param name="pasteClusterHandler"></param>
        /// <param name="appCache"></param>
        /// <param name="logger"></param>
        public StartHostedService(IServiceProvider serviceProvider, PasteClusterHandler pasteClusterHandler, IAppCache appCache, ILogger<StartHostedService> logger)
        {
            _serviceProvider = serviceProvider;
            _cluster_handler = pasteClusterHandler;
            _appCache = appCache;
            _logger = logger;
        }

        private const string node_list_cache_key = "cache:last:nodes";
        //private System.Timers.Timer _timer;

        /// <summary>
        /// 
        /// </summary>
        /// <param name="cancellationToken"></param>
        /// <returns></returns>
        /// <exception cref="System.NotImplementedException"></exception>
        public Task StartAsync(CancellationToken cancellationToken)
        {
            //可以从他处读取集群节点列表，然后把列表写入到系统中
            var reads = _appCache.Get<List<PasteNodeModel>>(node_list_cache_key);
            if (reads != null && reads.Count > 0)
            {
                foreach (var _node in reads)
                {
                    _cluster_handler.AddNodeToList(_node).Wait();
                    //await _cluster_handler.AddNodeToList(_node);
                }
            }
            //如果已知当前节点的Host(这里示例默认为80端口)信息，可以用下方的函数写入,也可以在启动的时候写入配置中
            // -e ClusterConfig:CurrentHost="http://192.168.1.100"
            //如果使用PasteSpider部署，则是-e ClusterConfig:CurrentHost="http://{{App.ProAddress}}"
            //_cluster_handler.Register("http://192.168.1.100",0,0);
            //启动集群中的当前节点(在这之前必须要确认当前节点的host是多少)
            _cluster_handler.StartCluster();
            //读取集群产生的数据，比如其他节点发送的数据等,这里一般是业务相关的消息
            ReadClusterChannel();

            //_timer = new System.Timers.Timer();
            //hit_index = new Random().Next(1,100);
            //_timer.Interval = 1000;
            //_timer.AutoReset = true;
            //_timer.Elapsed += _timer_Elapsed;
            //_timer.Start();
            return Task.CompletedTask;
        }

        //private int hit_index = 5;
        //测试产生数据进行交互
        //private void _timer_Elapsed(object sender, System.Timers.ElapsedEventArgs e)
        //{
        //    try
        //    {
        //        hit_index--;
        //        if (hit_index == 0)
        //        {
        //            _cluster_handler.PushMsgToNode($"{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")}");
        //            hit_index = new Random().Next(1, 20);
        //        }
        //    }
        //    catch (Exception exl)
        //    {
        //        _logger.LogException(exl);
        //    }
        //}

        /// <summary>
        /// 处理集群产生的数据
        /// </summary>
        private async Task ReadClusterChannel()
        {
                    try
            {
                var _info = await _cluster_handler.ChannelCluster.Reader.ReadAsync();
                if (_info != null && _info != default)
                {
                    switch (_info.msg_type)
                    {
                        case 1:
                            {
                                //当前节点成为master _info.body ==PasteMasterResponse.JsonString
                            }
                            break;
                        case 2:
                            {
                                //当前节点不再是master _info.body ==PasteMasterResponse.JsonString
                            }
                            break;
                        case 3:
                            {
                                //有节点加入 _info.body == PasteNodeModel.JsonString
                            }
                            break;
                        case 4:
                            {
                                if (_cluster_handler.CurrentIsMaster)
                                {
                                    var _node = System.Text.Json.JsonSerializer.Deserialize<PasteNodeModel>(_info.body);
                                    if (_node != null && _node != default)
                                    {
                                        if (!String.IsNullOrEmpty(_node.host))
                                        {
                                            await _appCache.HashDeleteAsync(node_list_cache_key, _node.host);
                                        }
                                    }
                                }
                            }
                            break;
                        default:
                            {
                                //业务代码 // _info.body是什么消息格式要基于业务而定
                            }
                            break;
                    }
                }
            }
            catch (Exception exl)
            {
                _logger.LogException(exl);
                await Task.Delay(1000);
            }
            finally
            {
                ReadClusterChannel();
            }
        }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="cancellationToken"></param>
        /// <returns></returns>
        /// <exception cref="System.NotImplementedException"></exception>
        public Task StopAsync(CancellationToken cancellationToken)
        {
            _cluster_handler.Dispose();
            return Task.CompletedTask;
        }
    }
}

pastecode.cn/PasteClusterv25.9.2

Get Started

Readme

组件适用

引入组件

开始使用

相关问答

逻辑介绍

写在最后

更新摘要

v24.07.28.03

v24.07.28.02

v24.07.28.01

v24.6.10.1

v24.6.09.1

v24.6.19.1

v24.6.21.1

25.08.001

Links

Keywords

Maintainers