AI 助理
文档备案控制台
开发者社区 开发与运维 文章 正文

面向连接的Socket Server的简单实现

2017-10-19 982
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介：

一、基本原理

有时候我们需要实现一个公共的模块，需要对多个其他的模块提供服务，最常用的方式就是实现一个Socket Server，接受客户的请求，并返回给客户结果。

这经常涉及到如果管理多个连接及如何多线程的提供服务的问题，常用的方式就是连接池和线程池，基本流程如下：

SocketServer

首先服务器端有一个监听线程，不断监听来自客户端的连接。

当一个客户端连接到监听线程后，便建立了一个新的连接。

监听线程将新建立的连接放入连接池进行管理，然后继续监听新来的连接。

线程池中有多个服务线程，每个线程都监听一个任务队列，一个建立的连接对应一个服务任务，当服务线程发现有新的任务的时候，便用此连接向客户端提供服务。

一个Socket Server所能够提供的连接数可配置，如果超过配置的个数则拒绝新的连接。

当服务线程完成服务的时候，客户端关闭连接，服务线程关闭连接，空闲并等待处理新的任务。

连接池的监控线程清除其中关闭的连接对象，从而可以建立新的连接。

二、对Socket的封装

Socket的调用主要包含以下的步骤：

clip_image001

调用比较复杂，我们首先区分两类Socket，一类是Listening Socket，一类是Connected Socket.

Listening Socket由MySocketServer负责，一旦accept，则生成一个Connected Socket，又MySocket负责。

MySocket主要实现的方法如下：

int MySocket::write(const char * buf, int length)
{
        int ret = 0;
        int left = length;
        int index = 0;
        while(left > 0)
        {
                ret = send(m_socket, buf + index, left, 0);
                if(ret == 0)
                        break;
                else if(ret == -1)
                {
                        break;
                }
                left -= ret;
                index += ret;
        }
        if(left > 0)
                return -1;
        return 0;
}

int MySocket::read(char * buf, int length)
{
        int ret = 0;
        int left = length;
        int index = 0;
        while(left > 0)
        {
                ret = recv(m_socket, buf + index, left, 0);
                if(ret == 0)
                        break;
                else if(ret == -1)
                        return -1;
                left -= ret;
                index += ret;
        }

        return index;
}

int MySocket::status()
{
        int status;
        int ret;
        fd_set checkset;
        struct timeval timeout;

        FD_ZERO(&checkset);
        FD_SET(m_socket, &checkset);

        timeout.tv_sec = 10;
        timeout.tv_usec = 0;

        status = select((int)m_socket + 1, &checkset, 0, 0, &timeout);
        if(status < 0)
                ret = -1;
        else if(status == 0)
                ret = 0;
        else
                ret = 0;
        return ret;
}

int MySocket::close()
{
        struct linger lin;
        lin.l_onoff = 1;
        lin.l_linger = 0;
        setsockopt(m_socket, SOL_SOCKET, SO_LINGER, (const char *)&lin, sizeof(lin));
        ::close(m_socket);
        return 0;
}

MySocketServer的主要方法实现如下：

int MySocketServer::init(int port)
{
        if((m_socket = socket(AF_INET, SOCK_STREAM, 0)) == -1)
        {
                return -1;
        }

        struct sockaddr_in serverAddr;
        memset(&serverAddr, 0, sizeof(struct sockaddr_in));
        serverAddr.sin_addr.s_addr = htonl(INADDR_ANY);
        serverAddr.sin_family = AF_INET;
        serverAddr.sin_port = htons(port);

        if(bind(m_socket, (struct sockaddr *)&serverAddr, sizeof(serverAddr)) == -1)
        {
                ::close(m_socket);
                return -1;
        }

        if(listen(m_socket, SOMAXCONN) == -1)
        {
                ::close(m_socket);
                return -1;
        }

        struct linger lin;
        lin.l_onoff = 1;
        lin.l_linger = 0;

        setsockopt(m_socket, SOL_SOCKET, SO_LINGER, (const char *)&lin, sizeof(lin));
        m_port = port;
        m_inited = true;
        return 0;
}

MySocket * MySocketServer::accept()
{
        int sock;
        struct sockaddr_in clientAddr;
        socklen_t clientAddrSize = sizeof(clientAddr);
        if((sock = ::accept(m_socket, (struct sockaddr *)&clientAddr, &clientAddrSize)) == -1)
        {
                return NULL;
        }
        MySocket* socket = new MySocket(sock);
        return socket;
}

MySocket * MySocketServer::accept(int timeout)
{
        struct timeval timeout;
        timeout.tv_sec = timeout;
        timeout.tv_usec = 0;

        fd_set checkset;
        FD_ZERO(&checkset);
        FD_SET(m_socket, &checkset);

        int status = (int)select((int)(m_socket + 1), &checkset, NULL, NULL, &timeout);
        if(status < 0)
                return NULL;
        else if(status == 0)
                return NULL;

        if(FD_ISSET(m_socket, &checkset))
        {
                return accept();
        }
}

三、线程池的实现

一个线程池一般有一个任务队列，启动的各个线程从任务队列中竞争任务，得到的线程则进行处理：list<MyTask *>  m_taskQueue;

任务队列由锁保护，使得线程安全：pthread_mutex_t m_queueMutex

任务队列需要条件变量来支持生产者消费者模式：pthread_cond_t m_cond

如果任务列表为空，则线程等待，等待中的线程个数为：m_numWaitThreads

需要一个列表来维护线程池中的线程：vector<MyThread *> m_threads

每个线程需要一个线程运行函数：

void * __thread_new_proc(void *p)
{
    ((MyThread *)p)->run();
    return 0;
}

每个线程由MyThread类负责，主要函数如下：

int MyThread::start()
{

    pthread_attr_t  attr;
    pthread_attr_init(&attr);
    pthread_attr_setschedpolicy(&attr, SCHED_FIFO);

    int ret = pthread_create(&m_thread, &attr, thread_func, args);
    pthread_attr_destroy(&attr);

    if(ret != 0)
        return –1;

}

int MyThread::stop()
{

    int ret = pthread_kill(m_thread, SIGINT);

    if(ret != 0)
        return –1;
}

int MyThread::join()

{

    int ret = pthread_join(m_thread, NULL);

    if(ret != 0)

        return –1;

}

void MyThread::run()

{

    while (false == m_bStop)

    {

        MyTask *pTask = m_threadPool->getNextTask();

        if (NULL != pTask)

        {

            pTask->process();

        }

    }

}

线程池由MyThreadPool负责，主要函数如下：

int MyThreadPool::init()
{

    pthread_condattr_t cond_attr;
    pthread_condattr_init(&cond_attr);
    pthread_condattr_setpshared(&cond_attr, PTHREAD_PROCESS_SHARED);
    int ret =  pthread_cond_init(&m_cond, &cond_attr);
    pthread_condattr_destroy(&cond_attr);

    if (ret_val != 0)
        return –1;

    pthread_mutexattr_t attr;
    pthread_mutexattr_init(&attr);
    pthread_mutexattr_setpshared(&attr, PTHREAD_PROCESS_SHARED);
    ret = pthread_mutex_init(&m_queueMutex, &attr);
    pthread_mutexattr_destroy(&attr);

    if (ret_val != 0)
        return –1;

    for (int i = 0; i< m_poolSize; ++i)
    {
        MyThread *thread = new MyThread(i+1, this);        
        m_threads.push_back(thread);
    }

    return 0;
}

int MyThreadPool::start()
{
    int ret;
    for (int i = 0; i< m_poolSize; ++i)
    {        
       ret = m_threads[i]->start();
       if (ret != 0)
           break;       
    }

    ret = pthread_cond_broadcast(&m_cond);

    if(ret != 0)
        return –1;
    return 0;
}

void MyThreadPool::addTask(MyTask *ptask)
{
    if (NULL == ptask)
        return;

    pthread_mutex_lock(&m_queueMutex);

    m_taskQueue.push_back(ptask);       

    if (m_waitingThreadCount > 0)
        pthread_cond_signal(&m_cond);

    pthread_mutex_unlock(&m_queueMutex);
}

MyTask * MyThreadPool::getNextTask()
{
    MyTask *pTask = NULL; 

    pthread_mutex_lock(&m_queueMutex);

    while (m_taskQueue.begin() == m_taskQueue.end())
    {  
        ++m_waitingThreadCount;

        pthread_cond_wait(&n_cond, &m_queueMutex);

        --m_waitingThreadCount;       
    }   

    pTask = m_taskQueue.front();

    m_taskQueue.pop_front();

    pthread_mutex_unlock(&m_queueMutex);

    return pTask;   
}

其中每一个任务的执行由MyTask负责，其主要方法如下：

void MyTask::process()

{

    //用read从客户端读取指令

    //对指令进行处理

    //用write向客户端写入结果

}

 

四、连接池的实现

每个连接池保存一个链表保存已经建立的连接：list<MyConnection *> * m_connections

当然这个链表也需要锁来进行多线程保护：pthread_mutex_t m_connectionMutex;

此处一个MyConnection也是一个MyTask，由一个线程来负责。

线程池也作为连接池的成员变量：MyThreadPool * m_threadPool   

连接池由类MyConnectionPool负责，其主要函数如下：

void MyConnectionPool::addConnection(MyConnection * pConn)
{

    pthread_mutex_lock(&m_connectionMutex);

    m_connections->push_back(pConn);

    pthread_mutex_unlock(&m_connectionMutex);

    m_threadPool->addTask(pConn);
}

MyConnectionPool也要启动一个背后的线程，来管理这些连接，移除结束的连接和错误的连接。

void MyConnectionPool::managePool()
{

    pthread_mutex_lock(&m_connectionMutex);

    for (list<MyConnection *>::iterator itr = m_connections->begin(); itr!=m_connections->end(); )
    {
        MyConnection *conn = *itr;        
        if (conn->isFinish())
        {
            delete conn;
            conn = NULL;
            list<MyConnection *>::iterator pos = itr++;
            m_connections->erase(pos);                         
        }
        else if (conn->isError())
        {

            //处理错误的连接
            ++itr;
        }
        else
        {
            ++itr;
        }
    }

    pthread_mutex_unlock(&m_connectionMutex);

}

 

五、监听线程的实现

监听线程需要有一个MySocketServer来监听客户端的连接，每当形成一个新的连接，查看是否超过设置的最大连接数，如果超过则关闭连接，如果未超过设置的最大连接数，则形成一个新的MyConnection，将其加入连接池和线程池。

MySocketServer *pServer = new MySocketServer(port);

MyConnectionPool *pPool = new MyConnectionPool();

while (!stopFlag)

{

    MySocket * sock = pServer->acceptConnection(5);

    if(sock != null)

    {

        if(m_connections.size > maxConnectionSize)

        {

            sock.close();

        }

        MyTask *pTask = new MyConnection();

        pPool->addConnection(pTask);      

    }

}

 

摘自:http://www.cnblogs.com/forfuture1978/archive/2010/09/12/1824443.html


本文转自茄子_2008，原文链接：http://www.cnblogs.com/xd502djj/archive/2010/09/16/1827676.html，如需转载请自行联系原作者

文章标签：
Java
监控
安全
关键词：
连接socket
socket server
长征2号
目录
相关文章
游客moiomvrp3vyac2
|
网络协议 开发者 Python
Python网络编程与Socket通信：连接世界的无限可能
在当今数字化时代，Python作为一种强大的编程语言，通过网络编程与Socket通信为我们打开了连接世界的无限可能。本文将深入探讨Python网络编程的基础知识、Socket通信的原理以及实际应用，帮助读者更好地理解并运用这一技术。
游客moiomvrp3vyac2
209 1
1941623231718325
|
网络协议 安全 Python
python中socket客户端关闭连接
【4月更文挑战第7天】本教程介绍了如何在TCP客户端中正确关闭连接。使用`close()`方法可关闭Socket连接并释放资源，示例代码显示了在正常和异常情况下关闭连接的方法。注意异常处理以确保在任何情况下都能关闭连接，并避免并发操作同一Socket，以保证连接的稳定和安全。掌握这些技巧对编写健壮的TCP客户端至关重要。
1941623231718325
1451 2
游客4ajxjr2o23cj2
|
监控 API 开发工具
Socket.IO介绍，以及怎么连接测试Socket.IO接口？
Socket.IO 是一个用于浏览器和服务器间实时双向通信的库，支持低延迟消息传递、跨平台运行及自动重连。文章介绍了其特点与调试需求，并详细说明如何使用 Apifox 工具创建、连接、发送/接收 Socket.IO 事件，以及团队协作和调试技巧。掌握这些技能可提升实时应用开发效率与质量。
游客4ajxjr2o23cj2
851 0
众所周知
|
Java Android开发
如何在Java中创建Socket连接？
如何在Java中创建Socket连接？
众所周知
304 0
愿天堂没有BUG（公众号同名）
|
网络协议 Unix Linux
腾讯三面：socket 优雅的关闭连接
close函数的作用是关闭套接字，并终止TCP连接。那终止连接之后，还在发送缓冲区的数据还能发送吗？接下来我们来看一下。
愿天堂没有BUG（公众号同名）
539 0
刘大猫.
|
关系型数据库 MySQL 数据库
docker启动mysql多实例连接报错Can’t connect to local MySQL server through socket ‘/var/run/mysqld/mysqld.sock’
docker启动mysql多实例连接报错Can’t connect to local MySQL server through socket ‘/var/run/mysqld/mysqld.sock’
刘大猫.
1471 0
众所周知
|
Java Android开发
Java Socket编程示例：服务器开启在8080端口监听，接收客户端连接并打印消息。
【6月更文挑战第23天】 Java Socket编程示例：服务器开启在8080端口监听，接收客户端连接并打印消息。客户端连接服务器，发送&quot;Hello, Server!&quot;后关闭。注意Android中需避免主线程进行网络操作。
众所周知
382 4
WCGhost-Netsafe
|
网络协议 Java
如何在Java中使用Socket编程实现TCP连接?
在Java中，通过Socket编程实现TCP连接非常常见。以下演示了基本的TCP通信流程，可根据具体需求进行扩展。
WCGhost-Netsafe
772 0
1941623231718325
|
网络协议 安全 Java
python中socket关闭连接和释放资源
【4月更文挑战第6天】本教程讲解了网络通信后如何正确关闭连接和释放资源，以保证程序稳定性、性能及避免资源泄露和安全问题。内容涵盖TCP、HTTP连接关闭，数据库连接释放，以及内存、文件句柄、线程等资源管理。建议使用异常处理、遵循编程语言最佳实践，并定期审查代码以优化资源管理。
1941623231718325
689 4
sumith
|
网络协议
逆向学习网络篇：通过Socket建立连接并传输数据
逆向学习网络篇：通过Socket建立连接并传输数据
sumith
355 0

热门文章

最新文章

  • 1
    2025年前端框架是该选vue还是react？有了大模型-例如通义灵码辅助编码，就不用纠结了！vue用的多选react，react用的多选vue
  • 2
    Linux使用ss命令查看socket状态
  • 3
    PyODPS 中使用 Python UDF
  • 4
    西门子S7-200 SMART全局变量和局部变量如何使用，编写带参数子程序并调用
  • 5
    运维编排场景系列---运行远端shell脚本
  • 6
    4行代码解决.git目录过大的问题
  • 7
    常用 Git 操作手册
  • 8
    Silverlight实用窍门系列：67.Silverlight下的Socket通讯
  • 9
    python3网络编程之socket
  • 10
    python实现两台不同主机之间进行通信（客户端和服务端）——Socket
  • 1
    Socket-TCP 上位机下位机数据交互框架
    309
  • 2
    基于C#实现的支持文件传输的Socket聊天室
    389
  • 3
    区分TCP/IP、HTTP、Socket三者的差异
    692
  • 4
    WebSocket、Socket、TCP 和 HTTP 的差别与应用场景
    2855
  • 5
    阿里云PAI-DLC任务Pytorch launch_agent Socket Timeout问题源码分析
    507
  • 6
    【YashanDB知识库】YAS-00402 failed to connect socket, errno 111
    183
  • 7
    【YashanDB知识库】查询空间数据提示This socket has been closed.
    195
  • 8
    【YashanDB知识库】YAS-00402 failed to connect socket, errno 111, error message "Connection refused"
    246
  • 9
    Socket如何实现客户端和服务器间的通信
    620
  • 10
    # 2个类轻松构建高效Socket通信库
    332

    • 相关电子书

    更多
  • 低代码开发师（初级）实战教程
  • 冬季实战营第三期：MySQL数据库进阶实战
  • 阿里巴巴DevOps 最佳实践手册
    • 下一篇
    云安全中心：病毒查杀