C++ session 管理方法

Question

我想创建一个 session 管理方案，简而言之，必须是这样的:

• 一个 std::map 来跟踪当前的事件 session (已填满带有 id 字符串和关联的消息队列)
• 每个消息队列上的一组线程。

有两种方法可以创建这个方案:

1. 将所有组件保留在主程序中，并在需要时使用新的 session ID 和关联的消息队列填充映射，并启动一个新的分离线程，并引用他将轮询的队列作为参数.

    //main program code
      _session_map[id] = queue;
      thread = new thread(&queue);
      thread.detach();
      thread.start();
    //main program code
   

2. 创建一个隐藏所有这些机制的 session_manager 类到主程序。

    class session_manager
    {
       //session_manager code
       std::string new_session(std::string id)
       {
          _session_map[id] = queue;
          thread = new thread(&queue);
          thread.detach();
          thread.start();
       }
   
       private:
         std::map<std::string,message_queue> _session_map;
    };
   

创建此方案的更好方法是什么？我不确定第二种方案是否可以正常工作，因为我不是很擅长使用线程。我也不知道如何跟踪关闭的 session ，有人有什么建议吗？

Answer 1

几年前我会给出与现在完全不同的答案。

我也不确定您所说的“ session 管理”是什么意思，或者队列与工作人员的关系。所以我将从一些假设开始:

• 您希望 N 个线程并行工作，相互竞争队列中的作业。
• 您的 session 是一种工作 session ，会一直持续到工作序列完成为止。

在过去的 C++ 线程编程时代，我实际上已经概述了关于如何实现这种方案的架构。

但我怀疑，您只需要完成一份工作，而不是一些“理论课”。

因此，我没有摆弄低级线程(特定于操作系统)，而是选择展示一种(也是特定于操作系统)更抽象的方式来完成任务。

优点:

• 并发编程无需处理锁、互斥锁、静态线程函数调用 pthis->Execute() 等。
• 概念上易于理解。消息 block 、源、目标、消息、工作对象(参与者/代理)。没有 promise future 的 C++-linq 响应式(Reactive)函数式编程替代尝试(尝试幽默!)。
• 看起来非常接近您的想法。

缺点:

• 隐藏了我们老人们为了解并度过了多年纯粹的快乐和绝望而感到自豪的所有低级内容。
• 很遗憾，只能在 Windows 平台上运行 (AFAIK)。

此处所有代码均使用 Windows 并发运行时。

#include "stdafx.h"
#include <thread>
#include <concrt.h>
#include <agents.h>
#include <iostream>

template<class _Job>
class Worker 
    : public concurrency::agent 
{
    concurrency::ISource<_Job> *m_source;
    volatile bool m_running;
    uint32_t m_counter;
public:
    Worker(concurrency::ISource<_Job> *source)
        : m_source(source)
        , m_running(true)
        , m_counter(0UL)
    {}
    ~Worker()
    {}
    uint32_t Counter() const
    {
        return m_counter;
    }
    void Stop()
    {
        m_running = false;
    }
    virtual void run()
    {
        while (m_running)
        {
            try
            {
                _Job job = concurrency::receive(m_source, 1000);
                m_counter++;
            }
            catch (concurrency::operation_timed_out& /*timeout*/)
            {
                std::cout << "Timeout." << std::endl;
            }
        }
        _Job job;
        while (concurrency::try_receive(m_source, job))
        {
            m_counter++;
        }
        done();
    }
};

typedef uint64_t Job_t;

int _tmain(int argc, _TCHAR* argv[])
{
    const size_t NUM_WORKERS = 4;
    concurrency::unbounded_buffer<Job_t> buffer;
    Worker<Job_t> workers[NUM_WORKERS] = 
        { Worker<Job_t>(&buffer)
        , Worker<Job_t>(&buffer)
        , Worker<Job_t>(&buffer)
        , Worker<Job_t>(&buffer)
        };
    std::vector<concurrency::agent*> agents;

    for (auto& worker : workers)
    {
        agents.push_back(&worker);
        worker.start();
    }

    for (uint64_t jobid = 0ULL; jobid < 1000000ULL; jobid++)
    { 
        concurrency::asend(buffer, jobid);
    }

    for (auto& worker : workers)
    {
        worker.Stop();
    }

    concurrency::agent::wait_for_all(NUM_WORKERS,&agents[0]);

    for (auto& worker : workers)
    {
        std::cout << "counter: " << worker.Counter() << std::endl;
    }

    return 0;
}