勤能補拙，Expter

成都游戲Coder，記錄游戲開發(fā)過程的筆記和心得！

導(dǎo)航

<

2009年8月

>

日

一

二

三

四

五

六

26

27

28

29

30

31

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

1

2

3

4

5

統(tǒng)計

隨筆 - 67
文章 - 1
評論 - 209
引用 - 0

留言簿(18)

隨筆分類

隨筆檔案

文章檔案

2008年10月 (1)

收藏夾

書庫(2) (rss)

程序人生

經(jīng)常去的論壇

csdn
vcbase
軍事網(wǎng)
sina軍事網(wǎng)

曾經(jīng)ACM

sicheng
wlgwstc
zzningxp大牛
百度fm
百度fm
中科院博士

積分與排名

積分 - 194068
排名 - 137

閱讀排行榜

評論排行榜

一個簡單線程池的實現(xiàn)

        以前寫線程池是在網(wǎng)絡(luò)編程的時候，一個線程池處理一個網(wǎng)絡(luò)套接字，隨著連接的增多，效率很低，
        最近主要是在封裝一個ipc程序(進程間通信機制) 主要
   涉及的技術(shù)：
    Winsock, 線程池
    因為光是基于線程池的技術(shù)效率還是很低，打算重新把其代碼整理重新封裝通過，在此基礎(chǔ)上通過完成端口來封裝一個簡單的高并發(fā)服務(wù)器。

    可能涉及的技術(shù)
    Winsock：   windows網(wǎng)絡(luò)通信
    完成端口:   Windows上服務(wù)器的大規(guī)模連接機制。
    線程池：    高效高利用率的線程機制。

本文主要實現(xiàn)一個線程池的例子，從基本原理入手，一個線程池會記錄每個線程的信息，以及每個線程的處理。
一般一個簡單線程池至少包含下列組成部分。

         1.線程池管理器（ThreadPoolManager）:用于創(chuàng)建并管理線程池
         2.工作線程（WorkThread）: 線程池中線程
         3.任務(wù)接口（Task）:每個任務(wù)必須實現(xiàn)的接口，以供工作線程調(diào)度任務(wù)的執(zhí)行。
         4.任務(wù)隊列:用于存放沒有處理的任務(wù)。提供一種緩沖機制。

1

#pragma once
2

3

#include "lock.h"
4

5

/// 線程池
6

namespace tp_ipc_peer_namespace
7

{
8

/// 接口
9

class task_object
10

{
11

public:
12

virtual ~task_object(){}
13

virtual void exec() = 0;
14

};
15

16

template< typename Functor>
17

class task
18

: public task_object
19

{
20

21

/// 禁止操作
22

private:
23

task( const task &);
24

task & operator =( const task & );
25

26

public:
27

typedef Functor functor_type;
28

29

task( const functor_type & functor)
30

: functor_( functor )
31

{ }
32

33

/// 執(zhí)行
34

virtual void exec()
35

{
36

functor_();
37

}
38

39

private:
40

Functor functor_;
41

42

};
43

44

class ctpool
45

{
46

typedef ctpool self_type;
47

48

public:
49

ctpool(void)
50

:tpool_running_( true )
51

{
52

_m_start_threads( 1 );
53

}
54

ctpool ( unsigned threadsize )
55

:tpool_running_(true)
56

{
57

_m_start_threads( threadsize );
58

}
59

60

template< typename Function>
61

void push( const Function & f)
62

{
63

/// 枷鎖
64

task_lock_.enter();
65

66

task_container_.push_back( new tp_ipc_peer_namespace::task<Function>( f ) );
67

68

task_lock_.leave();
69

70

}
71

72

~ctpool(void){}
73

74

private:
75

76

/// 創(chuàng)建線程池
77

void _m_start_threads( unsigned size )
78

{
79

if ( size == 0 )
80

size = 4;
81

82

for ( unsigned i = 0 ; i < size ; i++)
83

{
84

tinfo_type tinfo;
85

tinfo.state = 0;
86

tinfo.handle = (HANDLE)::_beginthreadex( 0 , 0 , _m_work_thread , NULL , NULL ,&(tinfo.tid) );
87

threads_.push_back( tinfo );
88

}
89

}
90

91

/// 喚醒
92

void _m_wakeup()
93

{
94

HANDLE handle = 0;
95

96

/// 對共享區(qū) 枷鎖
97

tlock_.enter();
98

std::vector<tinfo_type>::iterator it = threads_.begin(), end = threads_.end();
99

100

for ( ; it != end ; ++it )
101

{
102

if ( it->state == 0 ) /// 在等待狀態(tài)
103

{
104

handle = it->handle ;
105

it->state = 1;
106

break;
107

}
108

}
109

tlock_.leave();
110

111

while ( ::ResumeThread( handle ) != 1)
112

;
113

}
114

115

/// 掛起某個線程
116

void _m_suspend()
117

{
118

unsigned tid = ::GetCurrentThreadId();
119

HANDLE handle = 0;
120

121

tlock_.enter();
122

123

/// 對共享區(qū) 枷鎖
124

tlock_.enter();
125

std::vector<tinfo_type>::iterator it = threads_.begin(), end = threads_.end();
126

127

for ( ; it != end ; ++it )
128

{
129

if ( it->tid == tid ) /// 運行ID
130

{
131

handle = it->handle ;
132

it->state = 0;
133

break;
134

}
135

}
136

tlock_.leave();
137

138

/// 掛起
139

if ( handle)
140

{
141

::SuspendThread( handle );
142

}
143

}
144

145

/// 獲取task
146

tp_ipc_peer_namespace::task_object * _m_read_task()
147

{
148

while( tpool_running_ )
149

{
150

tp_ipc_peer_namespace::task_object * task = NULL;
151

152

/// 對共享區(qū) 枷鎖
153

task_lock_.enter();
154

if ( task_container_.size() )
155

{
156

task = *( task_container_.begin() );
157

task_container_.erase( task_container_.begin() );
158

}
159

task_lock_.leave();
160

161

if ( task )
162

{
163

return task;
164

}
165

else
166

_m_suspend();
167

}
168

return NULL;
169

}
170

171

private:
172

static unsigned __stdcall _m_work_thread(void * arg)
173

{
174

175

self_type & self = *reinterpret_cast<self_type*>(arg);
176

tp_ipc_peer_namespace::task_object * task = 0;
177

178

::SuspendThread(::GetCurrentThread());
179

180

while( true )
181

{
182

task = self._m_read_task();
183

if ( task )
184

{
185

task->exec();
186

187

delete task ;
188

task = 0;
189

}
190

else
191

break;
192

}
193

194

::_endthreadex( 0 );
195

return 0;
196

}
197

198

private:
199

/// 線程信息
200

struct tinfo_type
201

{
202

HANDLE handle;
203

unsigned tid;
204

unsigned long state; // 0 = sleep;
205

};
206

207

/// user define var
208

private:
209

/// 線程運行狀態(tài)
210

volatile bool tpool_running_;
211

/// 一個臨界區(qū)類
212

sync::csectionlock tlock_;
213

/// 線程信息
214

std::vector<tinfo_type> threads_;
215

///
216

sync::csectionlock task_lock_;
217

/// 一個回調(diào)函數(shù)
218

std::vector<task_object* > task_container_;
219

220

};
221

222

223

}

備注:在設(shè)計ipc的時候參考 http://man.chinaunix.net/tech/lyceum/linuxK/ipc/ipc.html
線程池設(shè)計 http://www.ibm.com/developerworks/cn/java/l-threadPool/

posted on 2009-08-09 18:10 expter 閱讀(4225) 評論(8) 編輯收藏引用

# re: 一個簡單線程池的實現(xiàn)[未登錄] 2009-08-10 01:57 fox

# re: 一個簡單線程池的實現(xiàn) 2009-08-10 09:19 abettor

# re: 一個簡單線程池的實現(xiàn)[未登錄] 2009-08-10 17:57 欲三更

# re: 一個簡單線程池的實現(xiàn) 2009-08-10 19:43 expter

# re: 一個簡單線程池的實現(xiàn) 2010-02-25 14:11 tcpcoder

# re: 一個簡單線程池的實現(xiàn) 2010-04-07 23:00 asuran

# re: 一個簡單線程池的實現(xiàn) 2011-10-26 17:47 wishper

# re: 一個簡單線程池的實現(xiàn) 2013-01-30 10:27 clement

青青草原综合久久大伊人导航_色综合久久天天综合_日日噜噜夜夜狠狠久久丁香五月_热久久这里只有精品