MultiExec.php
9.09 KB
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
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
<?php namespace common\components\redis\transaction;
use common\components\redis\IConnection;
use common\components\redis\PHPRedisConnection;
use common\components\redis\YIIRedisConnection;
/**
* Class MultiExec
* 注意:该类原型来自 \Predis\Transaction\MultiExec
* @package common\components\redis\transaction
*/
class MultiExec
{
private $state;
protected $connection;
protected $commands = [];
protected $attempts = 0;
protected $watchKeys = [];
protected $modeCAS = false;
/**
* @param IRedisConnection $connection Connection instance used by the transaction.
* @param array $options Initialization options.
*/
public function __construct(IConnection $connection, array $options = [])
{
$this->connection = clone $connection;
$this->state = new MultiExecState();
$this->configure($options ?: []);
$this->reset();
}
/**
* Configures the transaction using the provided options.
*
* @param IRedisConnection $connection Underlying connection instance.
* @param array $options Array of options for the transaction.
**/
protected function configure(array $options)
{
if (isset($options['cas'])) {
$this->modeCAS = (bool) $options['cas'];
}
if (isset($options['watch']) && $keys = $options['watch']) {
$this->watchKeys = $keys;
}
if (isset($options['retry'])) {
$this->attempts = (int) $options['retry'];
}
}
/**
* Resets the state of the transaction.
* 重置事务的状态
*/
protected function reset()
{
$this->state->reset();
$this->commands = [];
}
/**
* 初始化事务的上下文环境
*/
protected function initialize()
{
if ($this->state->isInitialized()) {
return;
}
if ($this->modeCAS) {
$this->state->flag(MultiExecState::CAS);
}
if ($this->watchKeys) {
$this->watch($this->watchKeys);
}
$cas = $this->state->isCAS();
$discarded = $this->state->isDiscarded();
if (!$cas || ($cas && $discarded)) {
$this->call('MULTI');
if ($discarded) {
$this->state->unflag(MultiExecState::CAS);
}
}
$this->state->unflag(MultiExecState::DISCARDED);
$this->state->flag(MultiExecState::INITIALIZED);
}
/**
* 代理 Connection 的 指令和方法,把REDIS操作捆绑到 MultiExec(事务类)
* $tr = new MultiExec();
* $tr->lpop();//lpop是REDIS命令
*
* 注意:调用该方法的命令会认为是事务的一部分,会被统计到$this->commands中去
*
* @param $method
* @param $arguments
* @return mixed
*/
public function __call($method, $arguments)
{
return $this->executeCommand($method, $arguments);
}
/**
* 执行REDIS命令
* 不调用初始化,提供给本类调用REDIS命令。
* @param string $method Command.
* @param array $arguments Arguments for the command.
*
* @return mixed
*/
protected function call($method, array $arguments = [])
{
return call_user_func_array([$this->connection, $method], $arguments);
}
/**
* 执行REDIS命令
* 调用初始化程序
* @param string $method Command.
* @param string $arguments 参数数组.
* @throws \Exception
* @return $this|mixed
*/
public function executeCommand($method, $arguments)
{
$this->initialize();
$response = call_user_func_array([$this->connection, $method], $arguments);
if ($this->state->isCAS()) {
return $response;
}
/**
* 开启 $tx->multi()之后,PHP REDI和YII REDIS在执行某条指令的时候返回的结果不同,
* YII REDIS是字符串之类的,PHP REDIS返回的是\Redis的实例。
*/
if($this->connection instanceof PHPRedisConnection && $response instanceof \Redis){
$this->commands[] = [
'method' => $method,
'arguments' => $arguments
];
} elseif($this->connection instanceof YIIRedisConnection && $response == 'QUEUED') {
$this->commands[] = [
'method' => $method,
'arguments' => $arguments
];
} else {
throw new \Exception('The server did not return a valid status response.');
}
}
/**
* Executes WATCH against one or more keys.
*
* @param string|array $keys One or more keys.
* @throws \Exception
* @return mixed
*/
public function watch($keys)
{
if ($this->state->isWatchAllowed()) {
throw new \Exception('Sending WATCH after MULTI is not allowed.');
}
$response = $this->call('WATCH', is_array($keys) ? $keys : array($keys));
$this->state->flag(MultiExecState::WATCH);
return $response;
}
/**
* Finalizes the transaction by executing MULTI on the server.
* @return MultiExec
*/
public function multi()
{
if ($this->state->check(MultiExecState::INITIALIZED | MultiExecState::CAS)) {
$this->state->unflag(MultiExecState::CAS);
$this->call('MULTI');
} else {
$this->initialize();
}
return $this;
}
/**
* Executes UNWATCH.
* @throws \Exception
* @return MultiExec
*/
public function unwatch()
{
$this->state->unflag(MultiExecState::WATCH);
$this->__call('UNWATCH', []);
return $this;
}
/**
* Resets the transaction by UNWATCH-ing the keys that are being WATCHed and
* DISCARD-ing pending commands that have been already sent to the server.
* @return MultiExec
*/
public function discard()
{
if ($this->state->isInitialized()) {
$this->call($this->state->isCAS() ? 'UNWATCH' : 'DISCARD');
$this->reset();
$this->state->flag(MultiExecState::DISCARDED);
}
return $this;
}
/**
* Executes the whole transaction.
* @return mixed
*/
public function exec()
{
return $this->execute();
}
/**
* Checks the state of the transaction before execution.
*
* @param mixed $callable Callback for execution.
* @throws \InvalidArgumentException
* @throws \Exception
*/
private function checkBeforeExecution($callable)
{
if ($this->state->isExecuting()) {
throw new \Exception('Cannot invoke "execute" or "exec" inside an active transaction context.');
}
if ($callable) {
if (!is_callable($callable)) {
throw new \InvalidArgumentException('The argument must be a callable object.');
}
if (!empty($this->commands)) {
$this->discard();
throw new \Exception('Cannot execute a transaction block after using fluent interface.');
}
} elseif ($this->attempts) {
$this->discard();
throw new \Exception('Automatic retries are supported only when a callable block is provided.');
}
}
/**
* Handles the actual execution of the whole transaction.
*
* @param mixed $callable Optional callback for execution.
* @throws AbortedMultiExecException
* @return array
*/
public function execute($callable = null)
{
$this->checkBeforeExecution($callable);
$execResponse = null;
$attempts = $this->attempts;
do {
if ($callable) {
$this->executeTransactionBlock($callable);
}
if (empty($this->commands)) {
if ($this->state->isWatching()) {
$this->discard();
}
return;
}
$execResponse = $this->call('EXEC');
if ($execResponse === null) {
if ($attempts === 0) {
throw new AbortedMultiExecException($this, 'The current transaction has been aborted by the server.');
}
$this->reset();
continue;
}
break;
} while ($attempts-- > 0);
$size = count($execResponse);
if ($size !== count($this->commands)) {
throw new \Exception('EXEC returned an unexpected number of response items.');
}
return $execResponse;
}
/**
* Passes the current transaction object to a callable block for execution.
* @param mixed $callable Callback.
* @throws null
*/
protected function executeTransactionBlock($callable)
{
$exception = null;
$this->state->flag(MultiExecState::INSIDEBLOCK);
try {
call_user_func($callable, $this);
} catch (\Exception $exception) {
$this->discard();
}
$this->state->unflag(MultiExecState::INSIDEBLOCK);
if ($exception) {
throw $exception;
}
}
}