Q&A-Ray-20180710

Posted 2020-12-03 charleszheng

Q:

如果集群多个客户端订阅会不会重复接收消息？

A:

集群环境用，有另外一个参数。

NodeManager类没有在框架里面：

public interface INodeManager : IGrainWithStringKey
{
    /// <summary>
    /// 初始化节点管理器
    /// </summary>
    /// <param name="nodeList"></param>
    /// <returns></returns>
    Task Init(List<string> nodeList);
    /// <summary>
    /// 获取一个节点名称作为当前节点的名称
    /// </summary>
    /// <returns></returns>
    Task<string> GetNodeName();
    /// <summary>
    /// 保持节点活动状态(超时节点会被回收)
    /// </summary>
    /// <param name="nodeName"></param>
    /// <returns>是否续活成功</returns>
    Task<bool> KeepLive(string nodeName);
}

public class NodeManager : Orleans.Grain, INodeManager
{
    List<string> nodeList;
    Dictionary<string, DateTime> nodeUserDict = new Dictionary<string, DateTime>();
    public Task Init(List<string> nodeList)
    {
        this.nodeList = nodeList;
        return Task.CompletedTask;
    }
    public Task<string> GetNodeName()
    {
        foreach (var node in nodeList)
        {
            if (nodeUserDict.TryGetValue(node, out var liveTime))
            {
                if (DateTime.UtcNow.Subtract(liveTime).TotalSeconds <= 35)
                    continue;
                else
                {
                    nodeUserDict[node] = DateTime.UtcNow;
                }
            }
            if (!nodeUserDict.ContainsKey(node))
            {
                nodeUserDict.Add(node, DateTime.UtcNow);
            }
            return Task.FromResult(node);
        }
        return Task.FromResult(string.Empty);
    }

    public Task<bool> KeepLive(string nodeName)
    {
        if (nodeUserDict.ContainsKey(nodeName))
        {
            if (nodeList.Contains(nodeName))
            {
                nodeUserDict[nodeName] = DateTime.UtcNow;
                return Task.FromResult(true);
            }
            else
                nodeUserDict.Remove(nodeName);
        }
        return Task.FromResult(false);
    }
}

Q:

没太搞明白这个怎么用，也就是集群客户端订阅会重复消费吗？

A:

这个类似分布式锁,设置只能启动三个节点,每次启动先去获取一个NodeName,如果获取得到就能启动，获取不到重试几次还不行就直接报错

Q:

就是在订阅的时候，限制启动节点。

---

示例：

var socketNodeManager=GrainClient.GrainFactory.GetGrain<INodeManager>("coin_websocket");
var nodeList=new List<string>(){ "Coin","Coin1","Coin2"};
socketNodeManager.Init(nodeList).GetAwaiter().GetResult();
while(true)
{
    var nodeName=socketNodeManager.GetNodeName().GetAwaiter().GetResult();
    if(!string.IsNullOrEmpty(nodeName))
    {
        Task.Factory.StartNew(async ()=>{
            while(true)
            {
               await Task.Delay(TimeSpan.FromSeconds(30));
               if (!await socketNodeManager.KeepLive(nodeName))
                {
                    await socketNodeManager.Init(nodeList);
                }
            }
        });
        SubscribeManage.Start(nodeName).GetAwaiter().GetResult();
        //或：SubscribeManage.Start(nodeName, nodeList).GetAwaiter().GetResult();
        break;
     }else
    {
        Task.Delay(20).Wait();
    }
}

?