beanstalkd-php demo(非扩展)

<?php 
require __DIR__ . '/Client.php';
$client = new Beanstalk\Client(['host' => '192.168.22.190']);
//
if (!$client->connect()) {
    echo 'connect error';exit;
} 
//
$client->useTube('test');
//
for($i=0; $i<100; ++$i) {
    $data= array(
        'function' => 'myEcho',
        'params' => array('hello world' . $i)
    );
    //优先级1024，延迟秒数2，超时秒数3，具体内容
    $client->put(1024, 2, 3, json_encode($data));
}
$client->disconnect();
exit('put over');

<?php

require __DIR__ . '/Client.php';

$client = new Beanstalk\Client(['host' => '192.168.22.190']);

if (!$client->connect()) {

echo 'connect error';exit;

}

$client->useTube('test');

for($i=0; $i<100; ++$i) {

$data= array(

'function' => 'myEcho',

'params' => array('hello world' . $i)

);

//优先级1024，延迟秒数2，超时秒数3，具体内容

$client->put(1024, 2, 3, json_encode($data));

}

$client->disconnect();

exit('put over');

<?php
//crontab每5分钟执行脚本
//检查脚本执行状态，如果正在执行，直接退出
checkRunning();
require __DIR__ . '/Client.php';
$task = new Beanstalk\Client(['host' => '192.168.22.190']);
//
if (!$task->connect()) {
 echo 'connect error';exit;
}
//
$task->watch('test');
//
while ($job = $task->reserve(30)) {   //设置超时时间为30秒
    $body = json_decode($job['body'], true);
    if (json_last_error()) {
        $task->bury($job['id']);  //保留任务，检查参数
    }
    $rs = call_user_func_array($body['function'], $body['params']);
    if ($rs) {
        $task->delete($job['id']);  //执行完成，彻底删除
    } else {
        $task->bury($job['id']);  //保留任务，可以让维护者 kick 回队列
    }
}
$task->disconnect();
file_put_contents(__DIR__ . '/check.txt', '1');
//
function myEcho($data){
 echo $data;
 echo "\n";
 return true;
}
//
function checkRunning() {
    $flag = file_get_contents(__DIR__ . '/check.txt');
    if ($flag == 1) {
        file_put_contents(__DIR__ . '/check.txt', '0');
    } else {
        exit;
    }
}

<?php

//crontab每5分钟执行脚本

//检查脚本执行状态，如果正在执行，直接退出

checkRunning();

require __DIR__ . '/Client.php';

$task = new Beanstalk\Client(['host' => '192.168.22.190']);

if (!$task->connect()) {

echo 'connect error';exit;

}

$task->watch('test');

while ($job = $task->reserve(30)) { //设置超时时间为30秒

$body = json_decode($job['body'], true);

if (json_last_error()) {

$task->bury($job['id']); //保留任务，检查参数

}

$rs = call_user_func_array($body['function'], $body['params']);

if ($rs) {

$task->delete($job['id']); //执行完成，彻底删除

} else {

$task->bury($job['id']); //保留任务，可以让维护者 kick 回队列

}

$task->disconnect();

file_put_contents(__DIR__ . '/check.txt', '1');

function myEcho($data){

echo $data;

echo "\n";

return true;

}

function checkRunning() {

$flag = file_get_contents(__DIR__ . '/check.txt');

if ($flag == 1) {

file_put_contents(__DIR__ . '/check.txt', '0');

} else {

exit;

}

<?php
//下载地址：https://github.com/davidpersson/beanstalk
/**
 * beanstalk: A minimalistic PHP beanstalk client.
 *
 * Copyright (c) 2009-2015 David Persson
 *
 * Distributed under the terms of the MIT License.
 * Redistributions of files must retain the above copyright notice.
 */
namespace Beanstalk;

use RuntimeException;

/**
 * An interface to the beanstalk queue service. Implements the beanstalk
 * protocol spec 1.9. Where appropriate the documentation from the protocol
 * has been added to the docblocks in this class.
 *
 * @link https://github.com/kr/beanstalkd/blob/master/doc/protocol.txt
 */
class Client {

	/**
	 * 最低优先级
	 * @var integer
	 */
	const MIN_PRIORITY = 0;

	/**
	 * 最高优先级 2^32-1
	 * @var integer
	 */
	const MAX_PRIORITY = 4294967295;

	/**
	 * 连接状态
	 * @var boolean
	 */
	public $connected = false;

	/**
	 * 配置
	 * @var array
	 */
	protected $_config = [];

	/**
	 * 当前连接的句柄
	 * @var resource
	 */
	protected $_connection;

	/**
	 * Constructor.
	 *
	 * @param array $config An array of configuration values:
	 *        - `'persistent'`  是否是长连接
	 *        - `'host'`        主机地址
	 *        - `'port'`        端口
	 *        - `'timeout'`     视为超时的秒数
	 *        - `'logger'`      一个psr-3兼容记录器实例 
	 *
	 * @link https://github.com/php-fig/fig-standards/blob/master/accepted/PSR-3-logger-interface.md
	 * @return void
	 */
	public function __construct(array $config = []) {
		$defaults = [
			'persistent' => true,
			'host' => '127.0.0.1',
			'port' => 11300,
			'timeout' => 1,
			'logger' => null
		];
		$this->_config = $config + $defaults;
	}

	/**
	 * 析构函数，与服务器断开连接
	 *
	 * @return void
	 */
	public function __destruct() {
		$this->disconnect();
	}

	/**
	 * Initiates a socket connection to the beanstalk server. The resulting
	 * stream will not have any timeout set on it. Which means it can wait
	 * an unlimited amount of time until a packet becomes available. This
	 * is required for doing blocking reads.
	 *
	 * @see \Beanstalk\Client::$_connection
	 * @see \Beanstalk\Client::reserve()
	 * @return boolean `true` if the connection was established, `false` otherwise.
	 */
	public function connect() {
		if (isset($this->_connection)) {
			$this->disconnect();
		}

		$function = $this->_config['persistent'] ? 'pfsockopen' : 'fsockopen';
		$params = [$this->_config['host'], $this->_config['port'], &$errNum, &$errStr];

		if ($this->_config['timeout']) {
			$params[] = $this->_config['timeout'];
		}
		$this->_connection = @call_user_func_array($function, $params);

		if (!empty($errNum) || !empty($errStr)) {
			$this->_error("{$errNum}: {$errStr}");
		}

		$this->connected = is_resource($this->_connection);

		if ($this->connected) {
			stream_set_timeout($this->_connection, -1);
		}
		return $this->connected;
	}

	/**
	 * Closes the connection to the beanstalk server by first signaling
	 * that we want to quit then actually closing the socket connection.
	 *
	 * @return boolean `true` if diconnecting was successful.
	 */
	public function disconnect() {
		if (!is_resource($this->_connection)) {
			$this->connected = false;
		} else {
			$this->_write('quit');
			$this->connected = !fclose($this->_connection);

			if (!$this->connected) {
				$this->_connection = null;
			}
		}
		return !$this->connected;
	}

	/**
	 * Pushes an error message to the logger, when one is configured.
	 *
	 * @param string $message The error message.
	 * @return void
	 */
	protected function _error($message) {
		if ($this->_config['logger']) {
			$this->_config['logger']->error($message);
		}
	}

	/**
	 * Writes a packet to the socket. Prior to writing to the socket will
	 * check for availability of the connection.
	 *
	 * @param string $data
	 * @return integer|boolean number of written bytes or `false` on error.
	 */
	protected function _write($data) {
		if (!$this->connected) {
			$message = 'No connecting found while writing data to socket.';
			throw new RuntimeException($message);
		}

		$data .= "\r\n";
		return fwrite($this->_connection, $data, strlen($data));
	}

	/**
	 * Reads a packet from the socket. Prior to reading from the socket
	 * will check for availability of the connection.
	 *
	 * @param integer $length Number of bytes to read.
	 * @return string|boolean Data or `false` on error.
	 */
	protected function _read($length = null) {
		if (!$this->connected) {
			$message = 'No connection found while reading data from socket.';
			throw new RuntimeException($message);
		}
		if ($length) {
			if (feof($this->_connection)) {
				return false;
			}
			$data = stream_get_contents($this->_connection, $length + 2);
			$meta = stream_get_meta_data($this->_connection);

			if ($meta['timed_out']) {
				$message = 'Connection timed out while reading data from socket.';
				throw new RuntimeException($message);
			}
			$packet = rtrim($data, "\r\n");
		} else {
			$packet = stream_get_line($this->_connection, 16384, "\r\n");
		}
		return $packet;
	}

	/* Producer Commands */

	/**
	 * The `put` command is for any process that wants to insert a job into the queue.
	 *
	 * @param integer $pri 优先级 0-4294967295
	 * @param integer $delay 延迟秒数
	 * @param integer $ttr 允许工作的秒数（超时秒数）
	 * @param string $data The job body.
	 * @return integer|boolean `false` 失败返回false 其他返回jobid
	 */
	public function put($pri, $delay, $ttr, $data) {
		$this->_write(sprintf("put %d %d %d %d\r\n%s", $pri, $delay, $ttr, strlen($data), $data));
		$status = strtok($this->_read(), ' ');

		switch ($status) {
			case 'INSERTED':
			case 'BURIED':
				return (integer) strtok(' '); // job id
			case 'EXPECTED_CRLF':
			case 'JOB_TOO_BIG':
			default:
				$this->_error($status);
				return false;
		}
	}

	/**
	 * The `use` command is for producers. Subsequent put commands will put
	 * jobs into the tube specified by this command. If no use command has
	 * been issued, jobs will be put into the tube named `default`.
	 *
	 * @param string $tube 管道名称最多200字节 replace操作
	 * @return string|boolean `false` 失败返回false 成功返回管道名称
	 */
	public function useTube($tube) {
		$this->_write(sprintf('use %s', $tube));
		$status = strtok($this->_read(), ' ');

		switch ($status) {
			case 'USING':
				return strtok(' ');
			default:
				$this->_error($status);
				return false;
		}
	}

	/**
	 * Pause a tube delaying any new job in it being reserved for a given time.
	 *
	 * @param string $tube The name of the tube to pause.
	 * @param integer $delay Number of seconds to wait before reserving any more
	 *        jobs from the queue.
	 * @return boolean `false` on error otherwise `true`.
	 */
	public function pauseTube($tube, $delay) {
		$this->_write(sprintf('pause-tube %s %d', $tube, $delay));
		$status = strtok($this->_read(), ' ');

		switch ($status) {
			case 'PAUSED':
				return true;
			case 'NOT_FOUND':
			default:
				$this->_error($status);
				return false;
		}
	}

	/* Worker Commands */

	/**
	 * Reserve a job (with a timeout).
	 *
	 * @param integer $timeout If given specifies number of seconds to wait for
	 *        a job. `0` returns immediately.
	 * @return array|false `false` on error otherwise an array holding job id
	 *         and body.
	 */
	public function reserve($timeout = null) {
		if (isset($timeout)) {
			$this->_write(sprintf('reserve-with-timeout %d', $timeout));
		} else {
			$this->_write('reserve');
		}
		$status = strtok($this->_read(), ' ');

		switch ($status) {
			case 'RESERVED':
				return [
					'id' => (integer) strtok(' '),
					'body' => $this->_read((integer) strtok(' '))
				];
			case 'DEADLINE_SOON':
			case 'TIMED_OUT':
			default:
				$this->_error($status);
				return false;
		}
	}

	/**
	 * Removes a job from the server entirely.
	 *
	 * @param integer $id The id of the job.
	 * @return boolean `false` on error, `true` on success.
	 */
	public function delete($id) {
		$this->_write(sprintf('delete %d', $id));
		$status = $this->_read();

		switch ($status) {
			case 'DELETED':
				return true;
			case 'NOT_FOUND':
			default:
				$this->_error($status);
				return false;
		}
	}

	/**
	 * Puts a reserved job back into the ready queue.
	 *
	 * @param integer $id The id of the job.
	 * @param integer $pri Priority to assign to the job.
	 * @param integer $delay Number of seconds to wait before putting the job in the ready queue.
	 * @return boolean `false` on error, `true` on success.
	 */
	public function release($id, $pri, $delay) {
		$this->_write(sprintf('release %d %d %d', $id, $pri, $delay));
		$status = $this->_read();

		switch ($status) {
			case 'RELEASED':
			case 'BURIED':
				return true;
			case 'NOT_FOUND':
			default:
				$this->_error($status);
				return false;
		}
	}

	/**
	 * Puts a job into the `buried` state Buried jobs are put into a FIFO
	 * linked list and will not be touched until a client kicks them.
	 *
	 * @param integer $id The id of the job.
	 * @param integer $pri *New* priority to assign to the job.
	 * @return boolean `false` on error, `true` on success.
	 */
	public function bury($id, $pri) {
		$this->_write(sprintf('bury %d %d', $id, $pri));
		$status = $this->_read();

		switch ($status) {
			case 'BURIED':
				return true;
			case 'NOT_FOUND':
			default:
				$this->_error($status);
				return false;
		}
	}

	/**
	 * Allows a worker to request more time to work on a job.
	 *
	 * @param integer $id The id of the job.
	 * @return boolean `false` on error, `true` on success.
	 */
	public function touch($id) {
		$this->_write(sprintf('touch %d', $id));
		$status = $this->_read();

		switch ($status) {
			case 'TOUCHED':
				return true;
			case 'NOT_TOUCHED':
			default:
				$this->_error($status);
				return false;
		}
	}

	/**
	 * Adds the named tube to the watch list for the current connection.
	 *
	 * @param string $tube Name of tube to watch.
	 * @return integer|boolean `false` on error otherwise number of tubes in watch list.
	 */
	public function watch($tube) {
		$this->_write(sprintf('watch %s', $tube));
		$status = strtok($this->_read(), ' ');

		switch ($status) {
			case 'WATCHING':
				return (integer) strtok(' ');
			default:
				$this->_error($status);
				return false;
		}
	}

	/**
	 * Remove the named tube from the watch list.
	 *
	 * @param string $tube Name of tube to ignore.
	 * @return integer|boolean `false` on error otherwise number of tubes in watch list.
	 */
	public function ignore($tube) {
		$this->_write(sprintf('ignore %s', $tube));
		$status = strtok($this->_read(), ' ');

		switch ($status) {
			case 'WATCHING':
				return (integer) strtok(' ');
			case 'NOT_IGNORED':
			default:
				$this->_error($status);
				return false;
		}
	}

	/* Other Commands */

	/**
	 * Inspect a job by its id.
	 *
	 * @param integer $id The id of the job.
	 * @return string|boolean `false` on error otherwise the body of the job.
	 */
	public function peek($id) {
		$this->_write(sprintf('peek %d', $id));
		return $this->_peekRead();
	}

	/**
	 * Inspect the next ready job.
	 *
	 * @return string|boolean `false` on error otherwise the body of the job.
	 */
	public function peekReady() {
		$this->_write('peek-ready');
		return $this->_peekRead();
	}

	/**
	 * Inspect the job with the shortest delay left.
	 *
	 * @return string|boolean `false` on error otherwise the body of the job.
	 */
	public function peekDelayed() {
		$this->_write('peek-delayed');
		return $this->_peekRead();
	}

	/**
	 * Inspect the next job in the list of buried jobs.
	 *
	 * @return string|boolean `false` on error otherwise the body of the job.
	 */
	public function peekBuried() {
		$this->_write('peek-buried');
		return $this->_peekRead();
	}

	/**
	 * Handles response for all peek methods.
	 *
	 * @return string|boolean `false` on error otherwise the body of the job.
	 */
	protected function _peekRead() {
		$status = strtok($this->_read(), ' ');

		switch ($status) {
			case 'FOUND':
				return [
					'id' => (integer) strtok(' '),
					'body' => $this->_read((integer) strtok(' '))
				];
			case 'NOT_FOUND':
			default:
				$this->_error($status);
				return false;
		}
	}

	/**
	 * Moves jobs into the ready queue (applies to the current tube).
	 *
	 * If there are buried jobs those get kicked only otherwise delayed
	 * jobs get kicked.
	 *
	 * @param integer $bound Upper bound on the number of jobs to kick.
	 * @return integer|boolean False on error otherwise number of jobs kicked.
	 */
	public function kick($bound) {
		$this->_write(sprintf('kick %d', $bound));
		$status = strtok($this->_read(), ' ');

		switch ($status) {
			case 'KICKED':
				return (integer) strtok(' ');
			default:
				$this->_error($status);
				return false;
		}
	}

	/**
	 * This is a variant of the kick command that operates with a single
	 * job identified by its job id. If the given job id exists and is in a
	 * buried or delayed state, it will be moved to the ready queue of the
	 * the same tube where it currently belongs.
	 *
	 * @param integer $id The job id.
	 * @return boolean `false` on error `true` otherwise.
	 */
	public function kickJob($id) {
		$this->_write(sprintf('kick-job %d', $id));
		$status = strtok($this->_read(), ' ');

		switch ($status) {
			case 'KICKED':
				return true;
			case 'NOT_FOUND':
			default:
				$this->_error($status);
				return false;
		}
	}

	/* Stats Commands */

	/**
	 * Gives statistical information about the specified job if it exists.
	 *
	 * @param integer $id The job id.
	 * @return string|boolean `false` on error otherwise a string with a yaml formatted dictionary.
	 */
	public function statsJob($id) {
		$this->_write(sprintf('stats-job %d', $id));
		return $this->_statsRead();
	}

	/**
	 * Gives statistical information about the specified tube if it exists.
	 *
	 * @param string $tube Name of the tube.
	 * @return string|boolean `false` on error otherwise a string with a yaml formatted dictionary.
	 */
	public function statsTube($tube) {
		$this->_write(sprintf('stats-tube %s', $tube));
		return $this->_statsRead();
	}

	/**
	 * Gives statistical information about the system as a whole.
	 *
	 * @return string|boolean `false` on error otherwise a string with a yaml formatted dictionary.
	 */
	public function stats() {
		$this->_write('stats');
		return $this->_statsRead();
	}

	/**
	 * Returns a list of all existing tubes.
	 *
	 * @return string|boolean `false` on error otherwise a string with a yaml formatted list.
	 */
	public function listTubes() {
		$this->_write('list-tubes');
		return $this->_statsRead();
	}

	/**
	 * Returns the tube currently being used by the producer.
	 *
	 * @return string|boolean `false` on error otherwise a string with the name of the tube.
	 */
	public function listTubeUsed() {
		$this->_write('list-tube-used');
		$status = strtok($this->_read(), ' ');

		switch ($status) {
			case 'USING':
				return strtok(' ');
			default:
				$this->_error($status);
				return false;
		}
	}

	/**
	 * Returns a list of tubes currently being watched by the worker.
	 *
	 * @return string|boolean `false` on error otherwise a string with a yaml formatted list.
	 */
	public function listTubesWatched() {
		$this->_write('list-tubes-watched');
		return $this->_statsRead();
	}

	/**
	 * Handles responses for all stat methods.
	 *
	 * @param boolean $decode Whether to decode data before returning it or not. Default is `true`.
	 * @return array|string|boolean `false` on error otherwise statistical data.
	 */
	protected function _statsRead($decode = true) {
		$status = strtok($this->_read(), ' ');

		switch ($status) {
			case 'OK':
				$data = $this->_read((integer) strtok(' '));
				return $decode ? $this->_decode($data) : $data;
			default:
				$this->_error($status);
				return false;
		}
	}

	/**
	 * Decodes YAML data. This is a super naive decoder which just works on
	 * a subset of YAML which is commonly returned by beanstalk.
	 *
	 * @param string $data The data in YAML format, can be either a list or a dictionary.
	 * @return array An (associative) array of the converted data.
	 */
	protected function _decode($data) {
		$data = array_slice(explode("\n", $data), 1);
		$result = [];

		foreach ($data as $key => $value) {
			if ($value[0] === '-') {
				$value = ltrim($value, '- ');
			} elseif (strpos($value, ':') !== false) {
				list($key, $value) = explode(':', $value);
				$value = ltrim($value, ' ');
			}
			if (is_numeric($value)) {
				$value = (integer) $value == $value ? (integer) $value : (float) $value;
			}
			$result[$key] = $value;
		}
		return $result;
	}
}

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

335

336

337

338

339

340

341

342

343

344

345

346

347

348

349

350

351

352

353

354

355

356

357

358

359

360

361

362

363

364

365

366

367

368

369

370

371

372

373

374

375

376

377

378

379

380

381

382

383

384

385

386

387

388

389

390

391

392

393

394

395

396

397

398

399

400

401

402

403

404

405

406

407

408

409

410

411

412

413

414

415

416

417

418

419

420

421

422

423

424

425

426

427

428

429

430

431

432

433

434

435

436

437

438

439

440

441

442

443

444

445

446

447

448

449

450

451

452

453

454

455

456

457

458

459

460

461

462

463

464

465

466

467

468

469

470

471

472

473

474

475

476

477

478

479

480

481

482

483

484

485

486

487

488

489

490

491

492

493

494

495

496

497

498

499

500

501

502

503

504

505

506

507

508

509

510

511

512

513

514

515

516

517

518

519

520

521

522

523

524

525

526

527

528

529

530

531

532

533

534

535

536

537

538

539

540

541

542

543

544

545

546

547

548

549

550

551

552

553

554

555

556

557

558

559

560

561

562

563

564

565

566

567

568

569

570

571

572

573

574

575

576

577

578

579

580

581

582

583

584

585

586

587

588

589

590

591

592

593

594

595

596

597

598

599

600

601

602

603

604

605

606

607

608

609

610

611

612

613

614

615

616

617

618

619

620

621

622

623

624

625

626

627

628

629

630

631

632

633

634

635

636

637

638

639

640

641

642

643

644

645

646

647

648

649

650

651

652

<?php

//下载地址：https://github.com/davidpersson/beanstalk

/**

* beanstalk: A minimalistic PHP beanstalk client.

* Distributed under the terms of the MIT License.

* Redistributions of files must retain the above copyright notice.

namespace Beanstalk;

use RuntimeException;

/**

* An interface to the beanstalk queue service. Implements the beanstalk

* protocol spec 1.9. Where appropriate the documentation from the protocol

* has been added to the docblocks in this class.

* @link https://github.com/kr/beanstalkd/blob/master/doc/protocol.txt

class Client {

/**

* 最低优先级

* @var integer

const MIN_PRIORITY = 0;

/**

* 最高优先级 2^32-1

* @var integer

const MAX_PRIORITY = 4294967295;

/**

* 连接状态

* @var boolean

public $connected = false;

/**

* 配置

* @var array

protected $_config = [];

/**

* 当前连接的句柄

* @var resource

protected $_connection;

/**

* Constructor.

* @param array $config An array of configuration values:

* - `'persistent'` 是否是长连接

* - `'host'` 主机地址

* - `'port'` 端口

* - `'timeout'` 视为超时的秒数

* - `'logger'` 一个psr-3兼容记录器实例

* @link https://github.com/php-fig/fig-standards/blob/master/accepted/PSR-3-logger-interface.md

* @return void

public function __construct(array $config = []) {

$defaults = [

'persistent' => true,

'host' => '127.0.0.1',

'port' => 11300,

'timeout' => 1,

'logger' => null

];

$this->_config = $config + $defaults;

}

/**

* 析构函数，与服务器断开连接

* @return void

public function __destruct() {

$this->disconnect();

}

/**

* Initiates a socket connection to the beanstalk server. The resulting

* stream will not have any timeout set on it. Which means it can wait

* an unlimited amount of time until a packet becomes available. This

* is required for doing blocking reads.

* @see \Beanstalk\Client::$_connection

* @see \Beanstalk\Client::reserve()

* @return boolean `true` if the connection was established, `false` otherwise.

public function connect() {

if (isset($this->_connection)) {

$this->disconnect();

}

$function = $this->_config['persistent'] ? 'pfsockopen' : 'fsockopen';

$params = [$this->_config['host'], $this->_config['port'], &$errNum, &$errStr];

if ($this->_config['timeout']) {

$params[] = $this->_config['timeout'];

}

$this->_connection = @call_user_func_array($function, $params);

if (!empty($errNum) || !empty($errStr)) {

$this->_error("{$errNum}: {$errStr}");

}

$this->connected = is_resource($this->_connection);

if ($this->connected) {

stream_set_timeout($this->_connection, -1);

}

return $this->connected;

}

/**

* Closes the connection to the beanstalk server by first signaling

* that we want to quit then actually closing the socket connection.

* @return boolean `true` if diconnecting was successful.

public function disconnect() {

if (!is_resource($this->_connection)) {

$this->connected = false;

} else {

$this->_write('quit');

$this->connected = !fclose($this->_connection);

if (!$this->connected) {

$this->_connection = null;

}

return !$this->connected;

}

/**

* Pushes an error message to the logger, when one is configured.

* @param string $message The error message.

* @return void

protected function _error($message) {

if ($this->_config['logger']) {

$this->_config['logger']->error($message);

}

/**

* Writes a packet to the socket. Prior to writing to the socket will

* check for availability of the connection.

* @param string $data

* @return integer|boolean number of written bytes or `false` on error.

protected function _write($data) {

if (!$this->connected) {

$message = 'No connecting found while writing data to socket.';

throw new RuntimeException($message);

}

$data .= "\r\n";

return fwrite($this->_connection, $data, strlen($data));

}

/**

* Reads a packet from the socket. Prior to reading from the socket

* will check for availability of the connection.

* @param integer $length Number of bytes to read.

* @return string|boolean Data or `false` on error.

protected function _read($length = null) {

if (!$this->connected) {

$message = 'No connection found while reading data from socket.';

throw new RuntimeException($message);

}

if ($length) {

if (feof($this->_connection)) {

return false;

}

$data = stream_get_contents($this->_connection, $length + 2);

$meta = stream_get_meta_data($this->_connection);

if ($meta['timed_out']) {

$message = 'Connection timed out while reading data from socket.';

throw new RuntimeException($message);

}

$packet = rtrim($data, "\r\n");

} else {

$packet = stream_get_line($this->_connection, 16384, "\r\n");

}

return $packet;

}

/* Producer Commands */

/**

* The `put` command is for any process that wants to insert a job into the queue.

* @param integer $pri 优先级 0-4294967295

* @param integer $delay 延迟秒数

* @param integer $ttr 允许工作的秒数（超时秒数）

* @param string $data The job body.

* @return integer|boolean `false` 失败返回false 其他返回jobid

public function put($pri, $delay, $ttr, $data) {

$this->_write(sprintf("put %d %d %d %d\r\n%s", $pri, $delay, $ttr, strlen($data), $data));

$status = strtok($this->_read(), ' ');

switch ($status) {

case 'INSERTED':

case 'BURIED':

return (integer) strtok(' '); // job id

case 'EXPECTED_CRLF':

case 'JOB_TOO_BIG':

default:

$this->_error($status);

return false;

}

/**

* The `use` command is for producers. Subsequent put commands will put

* jobs into the tube specified by this command. If no use command has

* been issued, jobs will be put into the tube named `default`.

* @param string $tube 管道名称最多200字节 replace操作

* @return string|boolean `false` 失败返回false 成功返回管道名称

public function useTube($tube) {

$this->_write(sprintf('use %s', $tube));

$status = strtok($this->_read(), ' ');

switch ($status) {

case 'USING':

return strtok(' ');

default:

$this->_error($status);

return false;

}

/**

* Pause a tube delaying any new job in it being reserved for a given time.

* @param string $tube The name of the tube to pause.

* @param integer $delay Number of seconds to wait before reserving any more

* jobs from the queue.

* @return boolean `false` on error otherwise `true`.

public function pauseTube($tube, $delay) {

$this->_write(sprintf('pause-tube %s %d', $tube, $delay));

$status = strtok($this->_read(), ' ');

switch ($status) {

case 'PAUSED':

return true;

case 'NOT_FOUND':

default:

$this->_error($status);

return false;

}

/* Worker Commands */

/**

* Reserve a job (with a timeout).

* @param integer $timeout If given specifies number of seconds to wait for

* a job. `0` returns immediately.

* @return array|false `false` on error otherwise an array holding job id

* and body.

public function reserve($timeout = null) {

if (isset($timeout)) {

$this->_write(sprintf('reserve-with-timeout %d', $timeout));

} else {

$this->_write('reserve');

}

$status = strtok($this->_read(), ' ');

switch ($status) {

case 'RESERVED':

return [

'id' => (integer) strtok(' '),

'body' => $this->_read((integer) strtok(' '))

];

case 'DEADLINE_SOON':

case 'TIMED_OUT':

default:

$this->_error($status);

return false;

}

/**

* Removes a job from the server entirely.

* @param integer $id The id of the job.

* @return boolean `false` on error, `true` on success.

public function delete($id) {

$this->_write(sprintf('delete %d', $id));

$status = $this->_read();

switch ($status) {

case 'DELETED':

return true;

case 'NOT_FOUND':

default:

$this->_error($status);

return false;

}

/**

* Puts a reserved job back into the ready queue.

* @param integer $id The id of the job.

* @param integer $pri Priority to assign to the job.

* @param integer $delay Number of seconds to wait before putting the job in the ready queue.

* @return boolean `false` on error, `true` on success.

public function release($id, $pri, $delay) {

$this->_write(sprintf('release %d %d %d', $id, $pri, $delay));

$status = $this->_read();

switch ($status) {

case 'RELEASED':

case 'BURIED':

return true;

case 'NOT_FOUND':

default:

$this->_error($status);

return false;

}

/**

* Puts a job into the `buried` state Buried jobs are put into a FIFO

* linked list and will not be touched until a client kicks them.

* @param integer $id The id of the job.

* @param integer $pri *New* priority to assign to the job.

* @return boolean `false` on error, `true` on success.

public function bury($id, $pri) {

$this->_write(sprintf('bury %d %d', $id, $pri));

$status = $this->_read();

switch ($status) {

case 'BURIED':

return true;

case 'NOT_FOUND':

default:

$this->_error($status);

return false;

}

/**

* Allows a worker to request more time to work on a job.

* @param integer $id The id of the job.

* @return boolean `false` on error, `true` on success.

public function touch($id) {

$this->_write(sprintf('touch %d', $id));

$status = $this->_read();

switch ($status) {

case 'TOUCHED':

return true;

case 'NOT_TOUCHED':

default:

$this->_error($status);

return false;

}

/**

* Adds the named tube to the watch list for the current connection.

* @param string $tube Name of tube to watch.

* @return integer|boolean `false` on error otherwise number of tubes in watch list.

public function watch($tube) {

$this->_write(sprintf('watch %s', $tube));

$status = strtok($this->_read(), ' ');

switch ($status) {

case 'WATCHING':

return (integer) strtok(' ');

default:

$this->_error($status);

return false;

}

/**

* Remove the named tube from the watch list.

* @param string $tube Name of tube to ignore.

* @return integer|boolean `false` on error otherwise number of tubes in watch list.

public function ignore($tube) {

$this->_write(sprintf('ignore %s', $tube));

$status = strtok($this->_read(), ' ');

switch ($status) {

case 'WATCHING':

return (integer) strtok(' ');

case 'NOT_IGNORED':

default:

$this->_error($status);

return false;

}

/* Other Commands */

/**

* Inspect a job by its id.

* @param integer $id The id of the job.

* @return string|boolean `false` on error otherwise the body of the job.

public function peek($id) {

$this->_write(sprintf('peek %d', $id));

return $this->_peekRead();

}

/**

* Inspect the next ready job.

* @return string|boolean `false` on error otherwise the body of the job.

public function peekReady() {

$this->_write('peek-ready');

return $this->_peekRead();

}

/**

* Inspect the job with the shortest delay left.

* @return string|boolean `false` on error otherwise the body of the job.

public function peekDelayed() {

$this->_write('peek-delayed');

return $this->_peekRead();

}

/**

* Inspect the next job in the list of buried jobs.

* @return string|boolean `false` on error otherwise the body of the job.

public function peekBuried() {

$this->_write('peek-buried');

return $this->_peekRead();

}

/**

* Handles response for all peek methods.

* @return string|boolean `false` on error otherwise the body of the job.

protected function _peekRead() {

$status = strtok($this->_read(), ' ');

switch ($status) {

case 'FOUND':

return [

'id' => (integer) strtok(' '),

'body' => $this->_read((integer) strtok(' '))

];

case 'NOT_FOUND':

default:

$this->_error($status);

return false;

}

/**

* Moves jobs into the ready queue (applies to the current tube).

* If there are buried jobs those get kicked only otherwise delayed

* jobs get kicked.

* @param integer $bound Upper bound on the number of jobs to kick.

* @return integer|boolean False on error otherwise number of jobs kicked.

public function kick($bound) {

$this->_write(sprintf('kick %d', $bound));

$status = strtok($this->_read(), ' ');

switch ($status) {

case 'KICKED':

return (integer) strtok(' ');

default:

$this->_error($status);

return false;

}

/**

* This is a variant of the kick command that operates with a single

* job identified by its job id. If the given job id exists and is in a

* buried or delayed state, it will be moved to the ready queue of the

* the same tube where it currently belongs.

* @param integer $id The job id.

* @return boolean `false` on error `true` otherwise.

public function kickJob($id) {

$this->_write(sprintf('kick-job %d', $id));

$status = strtok($this->_read(), ' ');

switch ($status) {

case 'KICKED':

return true;

case 'NOT_FOUND':

default:

$this->_error($status);

return false;

}

/* Stats Commands */

/**

* Gives statistical information about the specified job if it exists.

* @param integer $id The job id.

* @return string|boolean `false` on error otherwise a string with a yaml formatted dictionary.

public function statsJob($id) {

$this->_write(sprintf('stats-job %d', $id));

return $this->_statsRead();

}

/**

* Gives statistical information about the specified tube if it exists.

* @param string $tube Name of the tube.

* @return string|boolean `false` on error otherwise a string with a yaml formatted dictionary.

public function statsTube($tube) {

$this->_write(sprintf('stats-tube %s', $tube));

return $this->_statsRead();

}

/**

* Gives statistical information about the system as a whole.

* @return string|boolean `false` on error otherwise a string with a yaml formatted dictionary.

public function stats() {

$this->_write('stats');

return $this->_statsRead();

}

/**

* Returns a list of all existing tubes.

* @return string|boolean `false` on error otherwise a string with a yaml formatted list.

public function listTubes() {

$this->_write('list-tubes');

return $this->_statsRead();

}

/**

* Returns the tube currently being used by the producer.

* @return string|boolean `false` on error otherwise a string with the name of the tube.

public function listTubeUsed() {

$this->_write('list-tube-used');

$status = strtok($this->_read(), ' ');

switch ($status) {

case 'USING':

return strtok(' ');

default:

$this->_error($status);

return false;

}

/**

* Returns a list of tubes currently being watched by the worker.

* @return string|boolean `false` on error otherwise a string with a yaml formatted list.

public function listTubesWatched() {

$this->_write('list-tubes-watched');

return $this->_statsRead();

}

/**

* Handles responses for all stat methods.

* @param boolean $decode Whether to decode data before returning it or not. Default is `true`.

* @return array|string|boolean `false` on error otherwise statistical data.

protected function _statsRead($decode = true) {

$status = strtok($this->_read(), ' ');

switch ($status) {

case 'OK':

$data = $this->_read((integer) strtok(' '));

return $decode ? $this->_decode($data) : $data;

default:

$this->_error($status);

return false;

}

/**

* Decodes YAML data. This is a super naive decoder which just works on

* a subset of YAML which is commonly returned by beanstalk.

* @param string $data The data in YAML format, can be either a list or a dictionary.

* @return array An (associative) array of the converted data.

protected function _decode($data) {

$data = array_slice(explode("\n", $data), 1);

$result = [];

foreach ($data as $key => $value) {

if ($value[0] === '-') {

$value = ltrim($value, '- ');

} elseif (strpos($value, ':') !== false) {

list($key, $value) = explode(':', $value);

$value = ltrim($value, ' ');

}

if (is_numeric($value)) {

$value = (integer) $value == $value ? (integer) $value : (float) $value;

}

$result[$key] = $value;

}

return $result;

}

唐晓锋–BLOG

This is just a personal note.

发表评论取消回复

发表评论 取消回复

发表评论取消回复