Использование асинхронных сокетов.

Предлагаю вашему вниманию третью статью из цикла "Многопоточность" в PHP. В данной статье мы рассмотрим асинхронные или неблокирующие сокеты.

Обычно, когда мы открываем несколько сокетов, мы делаем это последовательно, т.е. открываем соединение к одному, ждем завершения соединения, что-либо делаем, закрываем, открываем следующий и т.д. Налицо неудобство такого подхода, когда ресурсов для соединения у нас много и с ними производятся одинаковые действия. Хорошо, если бы можно было не дожидаться завершения соединения, а сразу же открывать следущее. И такая возможность существует.

Но сначала давайте разберем само понятие "сокет". Что такое сокет?

Сокеты — это название программного интерфейса для обеспечения информационного обмена между процессами. Процессы при таком обмене могут исполняться как на одном компьютере, так и на различных, связанных между собой сетью. Сокет — абстрактный объект, представляющий конечную точку соединения.

В нашем случае сокет будет представлять собой соединение между нашей программой и компьютером, с которого требуется получить необходимую информацию.

Работа с сокетами достаточно проста:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

<?php
$url = 'www.mail.ru';
// создаем сокет
$sh = socket_create(AF_INET, SOCK_STREAM, SOL_TCP);
// определяем ip хоста
$ip = gethostbyname($url);
// открываем сокет
socket_connect($sh, $ip, '80');
// формируем http-заголовки
$headers  = "GET / HTTP/1.0\r\n";
$headers .= "Host: ".$url."\r\n";
$headers .= "User-Agent: Mozilla/5.0 (Windows; U; Windows NT 5.1; ru; rv:1.9.0.6) Gecko/2009011913 MRA 5.3 (build 02557) Firefox/3.0.6\r\n";
$headers .= "Accept: text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8\r\n";
$headers .= "Accept-Language: ru,en-us;q=0.7,en;q=0.3\r\n";
$headers .= "Accept-Charset: windows-1251,utf-8;q=0.7,*;q=0.7\r\n";
$headers .= "\r\n";
// записываем данные в сокет
socket_write($sh, $headers, strlen($headers));
// читаем данные из сокета
$result = '';
while ($r = socket_read($sh, 1024)) $result .= $r;
?>

Сначала функцией socket_create инициализируем сокет и задаем его характеристики – флаг AF_INET, переданный в качестве первого параметра, указывает, что создается сокет для интернета, второй параметр – тип сокета (самые распространенные – TCP и UDP) нам нужен TCP-сокет - SOCK_STREAM, и третий параметр задает протокол (TCP).
Затем с помощью функции socket_connect происходит соединение к указанному адресу и порту. Надо заметить, что эта функция принимает в качестве параметра не сам адрес хоста, а его ip адрес, поэтому сначала мы должны его определить функцией gethostbyname.
Теперь нам надо сформировать запрос, который мы отправим на сервер (подробнее о http-заголовках можно почитать в статье PHP и HTTP). Собственно отправку производит функция socket_write, второй параметр которой – это то, что мы отправляем и третий – длина отправляемой строки (т.е. всех сформированных заголовков).
Запись произведена, теперь функцией socket_read читаем данные, переданные для нас сервером.
В результате в переменной $result хранится хтмл, который нам необходим.

Задача у нас стоит та же, что и в предыдущих примерах: получить содержимое нескольких html-страниц. Вот код, реализующий это:

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

<?php
// страницы, содержимое которых надо получить
$urls = array('www.yandex.ru', 'www.google.ru', 'www.mail.ru', 'www.rambler.ru');
 
$rtasks = array(); // задачи чтения
$wtasks = array(); // задачи записи
$results = array(); // результаты
 
foreach ($urls as $url) {
	// открываем отдельный сокет
	$sh = socket_create(AF_INET, SOCK_STREAM, SOL_TCP);
	if (!$sh) continue;
 
	// таймаут для чтения
	socket_set_option($sh, SOL_SOCKET, SO_RCVTIMEO, array("sec" => 10, "usec" => 0));
	// таймаут для записи
	socket_set_option($sh, SOL_SOCKET, SO_SNDTIMEO, array("sec" => 10, "usec" => 0));
	// задаем неблокирующий режим сокетов
	socket_set_nonblock($sh);
	// определяем ip хоста
	$ip = gethostbyname($url);
	// соединяемся
	socket_connect($sh, $ip, 80);
 
	// добавляем в задачи для записи
	$wtasks[$url] = $sh;
}
 
// продолжаем, пока есть задачи для записи или чтения
while ($wtasks || $rtasks) {
	// массив для сокетов с возможностью чтения
	$rtasks_ = $rtasks;
	// массив для сокетов с возможностью записи
	$wtasks_ = $wtasks;
 
	// ждем результатов из сокетов
	$n = socket_select($rtasks_, $wtasks_, $e=null, 10);
	if ($n > 0) {
		// сокеты, доступные для записи
		foreach ($wtasks_ as $sh) {
			// ищем урл страницы по дескриптору сокета в массиве задач записи
			$url = array_search($sh, $wtasks);
			// удаляем из задач записи
			unset($wtasks[$url]);
			// добавляем в задачи чтения
			$rtasks[$url] = $sh;
			// формируем http-заголовки
			$headers  = "GET / HTTP/1.0\r\n";
			$headers .= "Host: ".$url."\r\n";
			$headers .= "User-Agent: Mozilla/5.0 (Windows; U; Windows NT 5.1; ru; rv:1.9.0.6) Gecko/2009011913 MRA 5.3 (build 02557) Firefox/3.0.6\r\n";
			$headers .= "Accept: text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8\r\n";
			$headers .= "Accept-Language: ru,en-us;q=0.7,en;q=0.3\r\n";
			$headers .= "Accept-Charset: windows-1251,utf-8;q=0.7,*;q=0.7\r\n";
			$headers .= "\r\n";
 
			// записываем в сокет
			if (socket_write($sh, $headers) === false) fclose($sh);
		}
 
		// сокеты, доступные для чтения
		foreach ($rtasks_ as $sh) {
			// ищем урл страницы по дескриптору сокета в массиве задач чтения
			$url = array_search($sh, $rtasks);
			if (!$url) continue;
			// считываем результат из сокета
			$result = '';
			while ($r = socket_read($sh, 1024)) $result .= $r;
			// закрываем сокет
			socket_close($sh);
			// удаляем из задач чтения
			unset($rtasks[$url]);
			// заносим html в массив результатов
			$results[$url] = $result;
		}
	}
	else {
		break;
	}
}
?>

Можно заметить, что код несильно отличается от кода для stream-функций.

В общих словах здесь происходит следующее. Каждый сокет инициализируется и переводится в неблокирующий режим. Затем в цикле производится проверка готовых к чтению или записи сокетов, и если такие существуют, производится соответствующая операция. Все действия протекают в асинхронном порядке, т.е не ждут окончания предыдущей операции. Например, если данные с мэйл.ру поступили раньше, чем от яндекса, то будут обрабатываться именно они, хотя коннект к яндексу мог произойти до коннекта к мэйлу. Можно сказать, что все происходит в несколько "потоков", когда параллельно ожидается готовность каждого сокета к какому-либо действию.

Теперь разберем код более подробно.
Итак, у нас есть массив с адресами страниц. В строке 9 запускается цикл для инициализации соединения с каждым хостом. В 11 строке создается сокет, затем в 15 и 17 строках мы задаем таймауты для чтения и записи соответственно. В 19 строчке функцией socket_set_nonblock задаем неблокирующий режим, превращающий каждый наш сокет в асинхронный. Это необходимо для того, чтобы после очередной операции над сокетом не происходило ожидание ее окончания, а продолжал работу остальной код программы. В 26 строке заносим инициализированный сокет в массив сокетов для записи.

Основной цикл начинается в 30 строчке. Он будет продолжаться до тех пор, пока у нас остаются сокеты, в которые нужно записать или из которых надо прочитать.
В 32 строке массиву $rtasks_ присваиваем массив сокетов, из которых должна быть прочитана информация. В 34-ой в $wtasks_ заносится массив с сокетами, в которые необходимо произвести запись.
Главная функция в этом цикле - socket_select. Она принимает на вход массивы сокетов, ожидающих чтения, ожидающих запись и массив для исключительных ситуаций, четвертый параметр, передаваемый в данную функцию – таймаут. Параметры-массивы передаются по ссылке, т.е когда произойдет ожидаемое событие (socket_select в этом случае вернет число больше 0), в них появятся соответствующие дескрипторы.
Если нашлись сокеты, доступные для записи, ищем урл, соответствующий каждому из готовых сокетов, и функцией socket_write записываем в него заголовки для получения хтмл страницы в ответе. Так же в 44-ой строке удаляем дескриптор из массива для записи, так как она будет произведена, и заносим его в строке 46 в массив для чтения, чтобы потом получить интересующий нас ответ.
Если же имеются сокеты, готовые нам ответить, считываем из них информацию функцией socket_read.
Profit! В массиве $results содержится вожделенный хтмл-код!

Замечание:
При вызове функции socket_connect, когда сокет переведен в неблокирующий режим функцией socket_set_nonblock, выдается предупреждение "Operation now in progress" ("Операция на незаблокированном сокете не может быть завершена немедленно"), а socket_connect возвращает false. Несмотря на это, код нормально отрабатывает.