场景

　　生产者和消费者队列, 生产者有多个, 消费者也有多个, 生产到消费需要异步.

下面用一个asp.netcore web-api项目来模拟

　　创建两个api, 一个get(), 一个set(), get返回一个字符串, set放入一个字符串, get返回的就是set进去的字符串.

　　实现如下:　　

[route("api/[controller]/[action]")]
public class foocontroller : control
{
    imessagequeue _mq;
    public foocontroller(imessagequeue mq)
    {
        _mq = mq;
    }

    [httpget]
    public string get()
    {
        string str = _mq.readone<string>();
        return str;
    }

    [httpget]
    public void set(string v)
    {
        _mq.writeone(v);
    }
}

public interface imessagequeue
{
    t readone<t>();
    void writeone<t>(t value);
}

public class messagequeue: imessagequeue
{
    private object _value;

    public t readone<t>()
    {
        return (t)_value;
    }

    public void writeone<t>(t value)
    {
        _value = value;

    }
}

接着在startup中把imessagequeue给注入了.

services.addsingleton<imessagequeue, messagequeue>();

运行后, 先调用/api/foo/set/?v=xxx, 再调用/api/foo/get/

可以看到成功返回了xxx

第二步, value字段改为队列:

使set进去的值不会被下一个覆盖, get取队列最前的值

为了线程安全, 这里使用了concurrentqueue<t>

代码如下:

public class messagequeue: imessagequeue
{
    private readonly concurrentqueue<object> _queue = new concurrentqueue<object>();

    public t readone<t>()
    {
        _queue.trydequeue(out object str);
        return (t)str ;
    }

    public void writeone<t>(tvalue)
    {
        _queue.enqueue(value);
    }
}

那么此时, 只要get不断地轮询, 就可以取到set生产出来的数据了.

调用/api/foo/set/

三, 异步阻塞

再增加需求, 调换get和set的顺序,先get后set模拟异步, (我这里的demo是个web-api会有http请求超时之类的…假装不存在)我想要get调用等待有数据时才返回.

也就是说我想要在浏览器地址栏输入http://localhost:5000/api/foo/get/之后会不断地转圈直到我用set接口放入一个值

方案a: while(true), 根本无情简直无敌, 死等read() != null时break; 为防单核满转加个thread.sleep();

方案b: monitor, 一个wait()一个exit/release();

但是以上两个方案都是基于thread的, .net4.0之后伴随concurrentqueue一起来的还有个blockingcollection<t>相当好用

方案c: 修改后代码如下:

public class messagequeue : imessagequeue
{
    private readonly blockingcollection<object> _queue = new blockingcollection<object>(new concurrentqueue<object>());

    public t readone<t>()
    {
        var obj = _queue.take();
        return (t)obj;
    }

    public void writeone<t>(t value)
    {
        _queue.add(value);
    }
}

此时, 如果先get, 会阻塞等待set; 如果已经有set过数据就会直接返回队列中的数据. get不会无功而返了. 基于这个类型, 可以实现更像样的订阅模型.

扩展rpc

这里的set是生产者, get是消费者, 那如果我的这个生产者并不单纯产生数据返回void而是需要等待一个结果的呢? 此时订阅模型不够用了, 我需要一个异步的rpc .

比如有个ask请求会携带参数发起请求, 并等待, 知道另外有个地方处理了这个任务产生结果, ask结束等待返回这个结果answer. 

我可以回头继续用方案a或b, 但连.net4.0都已经过去很久了, 所以应该用更好的基于task的异步方案.

代码如下, 首先新增两个接口:

public interface imessagequeue
{
    void respond<trequest, tresponse>(func<trequest, tresponse> func);
    task<tresponse> rpc<trequest, tresponse>(trequest req);

    t readone<t>();
    void writeone<t>(t data);
}

接着定义一个特殊的任务类:

public class rpctask<trequest, tresponse>
{
    public taskcompletionsource<tresponse> tcs { get; set; }
    public trequest request { get; set; }
}

实现刚才新加的两个接口:

public task<tresponse> rpc<trequest, tresponse>(trequest req)
{
    taskcompletionsource<tresponse> tcs = new taskcompletionsource<tresponse>();
    _queue.add(new rpctask<trequest, tresponse> { request = req, tcs = tcs});
    return tcs.task;
}

public void respond<trequest, tresponse>(func<trequest, tresponse> func)
{
    var obj = _queue.take();
    if(obj is rpctask<trequest, tresponse> t)
    {
        var response = func(t.request);
        t.tcs.setresult(response);
    }
}

同样的, 写两个web api接口, 一个请求等待结果 一个负责处理工作

[httpget]
public async task<string> ask(string v)
{
    var response = await _mq.rpc<myrequest, myresponse>(new myrequest { id = v });
    return $"[{response.donetime}] {response.id}";
}

[httpget]
public void answer()
{
    _mq.respond<myrequest, myresponse>((req)=> new myresponse { id = req.id, donetime = datetime.now });
}

上面还随便写了两个class作为请求和返回

public class myrequest
{
    public string id { get; set; }
}
public class myresponse
{
    public string id { get; set; }
    public datetime donetime { get; set; }
}

测试一下, 用浏览器或postman打开三个选项卡, 各发起一个ask接口的请求, 参数v分别为1 2 3, 三个选项卡都开始转圈等待

然后再打开一个选项卡访问answer接口, 处理刚才放进队列的任务, 发起一次之前的三个选项卡之中就有一个停止等待并显示返回数据. 需求实现.

这里用到的关键类型是taskcompletionsource<t>.

再扩展

如果是个分布式系统, 请求和处理逻辑不是在一个程序里呢? 那么这个队列可能也是一个单独的服务. 此时就要再加个返回队列了, 给队列中传输的每一个任务打上id, 返回队列中取出返回之后再找到id对于的tcs.setresult()