Служба мгновенных собщений своими руками

Все мы привыкли пользоваться аськой, многие этот функционал реализуют в своих проектах, кто-то использует БД, или сервер очередей, например memcacheq. Есть готовые решения, типа eJabber.

Если интересно, как можно сделать это самому, то wellcom под каст, где будет рассмотрена серверная часть «Службы мгновенных сообщений». С клиентской, я надеюсь, разберетесь сами...

Суть «Службы коротких сообщений — это в принципе и есть сервер очередей, работающий по протоколу HTTP, который легко должен интегрироваться с любым JS фреймворком. Вса результаты выдаются в JSON. Обмен с сервером осуществляется из браузера посредством AJAX.

по методу POST записываем данные по ключу, которым является url (или его часть). По методу GET извлекаем из очереди (по ключу, которым является url) то, что было записано. Итак имеем: одну Хештаблицу, ключом в которой являются uri, а данными — очереди. Очередь представляет строку собщения. При желании можно добавить и время. Хотя есть много идей по развитию.

итак ближе к теме:

1. #include <sys/types.h>
2. #include <sys/time.h>
3. #include <sys/queue.h>
4. #include <stdlib.h>
5. #include <err.h>
6. #include <event.h>
7. #include <evhttp.h>
8.  
9. #include <map>
10. #include <string>
11. #include <queue>
12.  
13. using namespace std;
14.  
15. std::map<string,queue<string> > ht;
16.  
17. void generic_handler(struct evhttp_request *req, void *arg)
18. {
19.     struct evbuffer *buf;
20.     buf = evbuffer_new();
21.     if (buf == NULL)
22.         err(1, "failed to create response buffer");
23.     string key = evhttp_request_uri(req);
24.     string out;
25.  
26.         if (req->type == EVHTTP_REQ_POST) {
27.  
28.             const char * str_len = evhttp_find_header(req->input_headers, "Content-Length");
29.             int len = atoi(str_len);
30.             out.assign((const char *)EVBUFFER_DATA(req->input_buffer),len);
31.             if ( ht.find(key) == ht.end() ) { 
32.                 queue<string> q;
33.                 q.push(out);
34.                 ht.insert( pair<string,queue<string> >(key,q));
35.             } else 
36.                 ht[key].push(out);
37.                 evbuffer_add_printf(buf, "{\"result\": \"Ok\"}\r\n");
38.             } else {
39.                 if ( ht.find(key) == ht.end() ) {
40.                 evbuffer_add_printf(buf, "{\"result\": null}\r\n" );
41.                  } else {
42.                     queue<string> q = ht[key];
43.                     if ( q.size() ) {
44.                         out = q.front();
45.                        q.pop();
46.                        ht[key]=q;
47.                        evbuffer_add_printf(buf, "{\"result\": \"%s\"}" , out.c_str() );
48.                 } else {
49.                 evbuffer_add_printf(buf, "{\"result\": null }" )
50.                 }
51.             }
52.         }
53.         evhttp_send_reply(req, HTTP_OK, "OK", buf);
54. }
55.  
56. int main(int argc, char **argv)
57. {
58.     struct evhttp *httpd;
59.     event_init();
60.     httpd = evhttp_start("0.0.0.0", 8080);
61.     evhttp_set_gencb(httpd, generic_handler, NULL);
62.     event_dispatch();
63.     evhttp_free(httpd);
64.     return 0;
65. }

немного пояснений по коду:
строки 58-63 инициализация WEB сервера. За основу взять WEB сервер основанный на libevent. У него отличная производительность. На моем ноуте 2.3ГГц он дает производительность 2к qps. Осуществляется обработка любого урла.
стр 20-22 — инициализирем буфер
стр 23 получаем REQUEST_IRI и используем для ключа. Есть много предложений по оптимизации.
стр 26 проверяем на POST. Несомненно можно еще проверить на HEAD (другие методы evhttp не поддерживает). Пока не будем усложнять жизнь.
стр 28-30 формируем переменную, в которой хранятся данные. Так как в буфере накапливается мусор, то мы записывает столько байт, сколько указано в заголовке Content-Length
стр 30-35 Если ключ такой не существует, то заводим новую очередь и вставляем в очередь элемент данных
стр 36 иначе просто вставляем в очередь элемент данных
стр 38 — отрабатывает метод GET
стр 39 проверяет существует ли ключ
стр 40 — нет — выводим сообщение о пустом результате
стр 42- получаем данные по ключу
стр 43 — проверяем пустая ли очередь,
стр 44-47 — нет, выбираем сообщение из очереди и выводим его, очередь уменьшается на одно сообщение
Можно немного пофлеймить, что надо сделать эскейпинг. Да, обязательно это добавлю.
стр 49 да, очередь пуста, сообщаем об этом.
стр 53 финализируем запрос, отправляем код ответа 200 OK

Необходимо отметить, что модель однопоточная, по этому ни каких блокировок на запись делать не надо. Хотя этот вопрос еще мной будет прорабатываться.

Результаты ab
Concurrency Level: 3
Time taken for tests: 0.415 seconds
Complete requests: 1000
Failed requests: 0
Write errors: 0
Total transferred: 83000 bytes
HTML transferred: 19000 bytes
Requests per second: 2409.31 [#/sec] (mean)
Time per request: 1.245 [ms] (mean)
Time per request: 0.415 [ms] (mean, across all concurrent requests)
Transfer rate: 195.29 [Kbytes/sec] received

объем потребляемой память на 10 000 сообщений — чуть более 600К

Вообще-то планируется поставить за nginx, пусть он отвечает за безопасность (ngx_http_accesskey_module)
кусок конфига:
location /test {
proxy_pass 127.0.0.1:8080;
## и еще директивы ngx_http_accesskey_module
}
но через nginx производительность доходит до 800 rps

Есть много идей — как и куда двигаться дальше. Например, отображение статусов активности, антиспам.
Любые иные идеи приветствуются. Пока сижу без работы, по этому нет проекта, на котором могу это опробовать. По моим расчетам одновременно около 10 тыс клиентов спокойно потянет, если делать запросы с частотой 1 -1.5 мин ( 130-200 rps).