Очереди сообщений в Linux: System V IPC и POSIX mqueue — полное руководство | AdminWiki
Logo

Очереди сообщений в Linux: System V IPC и POSIX mqueue — полное руководство

18 августа 2026 11 мин. чтения
Содержание статьи

Введение в очереди сообщений Linux

Очереди сообщений решают задачу обмена дискретными данными между процессами без привлечения внешних брокеров. Ядро Linux управляет очередью как буфером: один процесс помещает сообщение, другой извлекает его. В отличие от pipe, который передаёт неструктурированный поток байтов, очередь сохраняет границы сообщений и позволяет задавать приоритеты. В отличие от shared memory, не требуется отдельная синхронизация доступа: ядро само сериализует операции отправки и получения.

В Linux доступны две основные реализации: классическая System V IPC и современная POSIX mqueue. Первая появилась в Unix System V и использует функции msgget, msgsnd, msgrcv. Вторая определена стандартом POSIX.1-2001 и предоставляет функции mq_open, mq_send, mq_receive, а также механизм асинхронных уведомлений mq_notify. POSIX-интерфейс переносим между Unix-подобными системами и имеет файловое представление через mqueuefs.

Эта статья содержит практические примеры на языке C для обеих реализаций, разбор лимитов, типичных ошибок и критерии выбора под конкретную задачу.

System V IPC: очереди сообщений

System V IPC предоставляет три механизма межпроцессного взаимодействия: очереди сообщений, семафоры и разделяемую память. Очереди идентифицируются ключом типа key_t, а не именем. Это создаёт дополнительный шаг: нужно сгенерировать ключ или договориться о нём между процессами.

Создание очереди: msgget()

Функция msgget создаёт новую очередь или получает идентификатор существующей:

int msgget(key_t key, int msgflg);

Параметр key может принимать значение IPC_PRIVATE, которое всегда создаёт новую очередь. Для связи между независимыми процессами ключ генерируют через ftok() на основе пути к файлу и проекта. Флаги в msgflg: IPC_CREAT создаёт очередь, если её нет, IPC_EXCL вместе с IPC_CREAT вызывает ошибку, если очередь уже существует. Права доступа задаются так же, как для файлов: 0666, 0644.

#include 
#include 
#include 

int main() {
    key_t key = ftok("/tmp/msg_queue_example", 65);
    int msgid = msgget(key, IPC_CREAT | 0666);
    if (msgid == -1) {
        perror("msgget");
        exit(1);
    }
    printf("Очередь создана, msgid = %d\n", msgid);
    return 0;
}

При ошибке msgget возвращает -1, а errno содержит код ошибки: EACCES при недостатке прав, EEXIST при конфликте с IPC_EXCL, ENOSPC при исчерпании системного лимита на количество очередей.

Отправка сообщений: msgsnd()

Сообщение в System V очереди состоит из типа и данных. Тип задаётся первым полем структуры и должен быть положительным числом:

struct msgbuf {
    long mtype;
    char mtext[100];
};

Функция отправки:

int msgsnd(int msqid, const void *msgp, size_t msgsz, int msgflg);

Параметр msgsz - это размер данных, то есть размер структуры без учёта поля mtype. Флаг IPC_NOWAIT делает операцию неблокирующей: если очередь заполнена, функция вернёт -1 с errno = EAGAIN вместо ожидания освобождения места.

#include 
#include 
#include 

struct msgbuf {
    long mtype;
    char mtext[100];
};

int main() {
    key_t key = ftok("/tmp/msg_queue_example", 65);
    int msgid = msgget(key, 0666);
    if (msgid == -1) {
        perror("msgget");
        return 1;
    }

    struct msgbuf msg;
    msg.mtype = 1;
    strcpy(msg.mtext, "Привет от процесса-отправителя");

    if (msgsnd(msgid, &msg, strlen(msg.mtext) + 1, 0) == -1) {
        perror("msgsnd");
        return 1;
    }
    printf("Сообщение отправлено\n");
    return 0;
}

Размер данных не должен превышать системный лимит msgmax, иначе функция вернёт EINVAL.

Получение сообщений: msgrcv()

Функция получения сообщения позволяет выбирать, какое сообщение извлечь, по типу:

ssize_t msgrcv(int msqid, void *msgp, size_t msgsz, long msgtyp, int msgflg);

Параметр msgtyp управляет выборкой: значение 0 извлекает первое сообщение в очереди, положительное значение извлекает первое сообщение с указанным типом, отрицательное значение извлекает сообщение с минимальным типом, не превышающим модуль msgtyp. Флаг MSG_NOERROR позволяет обрезать сообщение до размера msgsz, если оно длиннее буфера. Без этого флага при нехватке места функция вернёт E2BIG.

#include 
#include 

struct msgbuf {
    long mtype;
    char mtext[100];
};

int main() {
    key_t key = ftok("/tmp/msg_queue_example", 65);
    int msgid = msgget(key, 0666);
    if (msgid == -1) {
        perror("msgget");
        return 1;
    }

    struct msgbuf msg;
    ssize_t n = msgrcv(msgid, &msg, sizeof(msg.mtext), 0, 0);
    if (n == -1) {
        perror("msgrcv");
        return 1;
    }
    printf("Получено сообщение типа %ld: %s\n", msg.mtype, msg.mtext);
    return 0;
}

При использовании IPC_NOWAIT и пустой очереди функция вернёт -1 с errno = ENOMSG.

Управление и удаление очереди: msgctl()

Функция msgctl управляет параметрами очереди и удаляет её:

int msgctl(int msqid, int cmd, struct msqid_ds *buf);

Команда IPC_STAT копирует информацию об очереди в структуру msqid_ds, IPC_SET обновляет параметры, IPC_RMID удаляет очередь немедленно. Удаление не зависит от того, есть ли процессы, которые держат очередь открытой: все последующие операции с этим msqid вернут EIDRM.

#include 
#include 

int main() {
    key_t key = ftok("/tmp/msg_queue_example", 65);
    int msgid = msgget(key, 0666);
    if (msgid == -1) {
        perror("msgget");
        return 1;
    }

    if (msgctl(msgid, IPC_RMID, NULL) == -1) {
        perror("msgctl");
        return 1;
    }
    printf("Очередь удалена\n");
    return 0;
}

Очистка очередей после завершения работы обязательна. Очереди System V существуют до явного удаления или перезагрузки системы, и каждая забытая очередь расходует память ядра.

POSIX mqueue: очереди сообщений

POSIX mqueue использует имена вместо числовых ключей. Имя очереди начинается со слеша, например /myqueue. Ядро хранит очереди в виртуальной файловой системе mqueuefs, которую можно смонтировать для просмотра и управления через файловые операции.

Создание и открытие очереди: mq_open()

Функция mq_open создаёт или открывает очередь по имени:

mqd_t mq_open(const char *name, int oflag, ...);

Флаги: O_CREAT создаёт очередь при отсутствии, O_EXCL с O_CREAT требует, чтобы очередь не существовала, O_RDONLY, O_WRONLY, O_RDWR задают режим доступа, O_NONBLOCK включает неблокирующий режим. При создании очереди дополнительно передаются права доступа и указатель на структуру mq_attr с атрибутами.

#include 
#include 
#include 

int main() {
    mqd_t mq = mq_open("/myqueue", O_CREAT | O_RDWR, 0666, NULL);
    if (mq == (mqd_t)-1) {
        perror("mq_open");
        exit(1);
    }
    printf("Очередь POSIX создана\n");
    mq_close(mq);
    return 0;
}

Если attr равен NULL, используются системные значения по умолчанию. При ошибке возвращается (mqd_t)-1, а errno содержит EACCES, EEXIST, ENOSPC или ENAMETOOLONG при слишком длинном имени.

Отправка сообщений: mq_send()

Функция отправки принимает приоритет сообщения:

int mq_send(mqd_t mqdes, const char *msg_ptr, size_t msg_len, unsigned int msg_prio);

Приоритет 0 - самый низкий. Сообщения с более высоким приоритетом извлекаются раньше сообщений с низким. В неблокирующем режиме при заполненной очереди функция вернёт -1 с errno = EAGAIN.

#include 
#include 
#include 

int main() {
    mqd_t mq = mq_open("/myqueue", O_WRONLY);
    if (mq == (mqd_t)-1) {
        perror("mq_open");
        return 1;
    }

    const char *text = "Сообщение через POSIX mqueue";
    if (mq_send(mq, text, strlen(text) + 1, 5) == -1) {
        perror("mq_send");
        return 1;
    }
    printf("Сообщение отправлено с приоритетом 5\n");
    mq_close(mq);
    return 0;
}

Длина сообщения не должна превышать атрибут mq_msgsize очереди. При превышении функция вернёт EMSGSIZE.

Получение сообщений: mq_receive()

Функция получения извлекает самое приоритетное сообщение и возвращает его приоритет через указатель:

ssize_t mq_receive(mqd_t mqdes, char *msg_ptr, size_t msg_len, unsigned int *msg_prio);

Если msg_prio не NULL, туда записывается приоритет полученного сообщения. Буфер должен быть не меньше mq_msgsize очереди, иначе функция вернёт EMSGSIZE. В неблокирующем режиме при пустой очереди возвращается -1 с errno = EAGAIN.

#include 
#include 

int main() {
    mqd_t mq = mq_open("/myqueue", O_RDONLY);
    if (mq == (mqd_t)-1) {
        perror("mq_open");
        return 1;
    }

    char buffer[8192];
    unsigned int prio;
    ssize_t n = mq_receive(mq, buffer, sizeof(buffer), &prio);
    if (n == -1) {
        perror("mq_receive");
        return 1;
    }
    printf("Получено сообщение с приоритетом %u: %s\n", prio, buffer);
    mq_close(mq);
    return 0;
}

mq_close закрывает дескриптор очереди в текущем процессе, но очередь продолжает существовать в системе. mq_unlink удаляет очередь по имени. После вызова mq_unlink очередь исчезает, когда все процессы закроют свои дескрипторы.

#include 
#include 

int main() {
    mqd_t mq = mq_open("/myqueue", O_RDWR);
    if (mq == (mqd_t)-1) {
        perror("mq_open");
        return 1;
    }

    mq_close(mq);
    if (mq_unlink("/myqueue") == -1) {
        perror("mq_unlink");
        return 1;
    }
    printf("Очередь закрыта и удалена\n");
    return 0;
}

Порядок важен: сначала mq_close для каждого открытого дескриптора, затем mq_unlink для удаления имени. Пропуск mq_unlink приводит к утечке очереди в системе.

Асинхронные уведомления: mq_notify()

Функция mq_notify регистрирует уведомление о появлении сообщения в пустой очереди. Это позволяет избежать блокировки или циклического опроса:

int mq_notify(mqd_t mqdes, const struct sigevent *sevp);

Можно зарегистрировать сигнал через SIGEV_SIGNAL или запуск потока через SIGEV_THREAD. Только один процесс может быть зарегистрирован на уведомление для конкретной очереди. После срабатывания уведомления регистрация снимается, и для повторного получения нужно вызвать mq_notify снова.

#include 
#include 
#include 
#include 

static void handler(int sig) {
    printf("Появилось сообщение в очереди\n");
}

int main() {
    mqd_t mq = mq_open("/myqueue", O_RDONLY | O_NONBLOCK);
    if (mq == (mqd_t)-1) {
        perror("mq_open");
        exit(1);
    }

    struct sigevent sev;
    sev.sigev_notify = SIGEV_SIGNAL;
    sev.sigev_signo = SIGUSR1;
    signal(SIGUSR1, handler);

    if (mq_notify(mq, &sev) == -1) {
        perror("mq_notify");
        exit(1);
    }

    while (1) {
        pause();
    }
    return 0;
}

Уведомление срабатывает только при переходе очереди из пустого состояния в непустое. Если в очереди уже есть сообщения, уведомление не придёт, пока очередь не опустеет и не получит новое сообщение.

Сравнение System V IPC и POSIX mqueue

Выбор между реализациями зависит от требований проекта. Таблица показывает ключевые различия:

КритерийSystem V IPCPOSIX mqueue
ИдентификацияЧисловой ключ key_tИмя, начинающееся с /
ПереносимостьUnix-подобные, есть нюансыPOSIX.1-2001, шире поддержка
Асинхронные уведомленияНетЕсть через mq_notify
Файловый интерфейсНетЧерез mqueuefs
Выбор сообщения при полученииПо типу mtypeПо приоритету
Удалениеmsgctl с IPC_RMIDmq_unlink

Для новых проектов предпочтительнее POSIX mqueue: интерфейс проще, есть уведомления, переносимость выше. System V IPC оправдан при поддержке legacy-кода, где уже используется этот механизм, или при необходимости выборки сообщений по типу, а не по приоритету.

Если задача выходит за пределы одного хоста и требует сетевого взаимодействия, стоит рассмотреть брокеры сообщений. Материал об основах очередей сообщений объясняет, когда нужен RabbitMQ или Kafka, а когда достаточно встроенных механизмов ядра.

Лимиты и настройка очередей сообщений

Системные лимиты ограничивают размер сообщений, количество сообщений в очереди и общее число очередей. Значения по умолчанию зависят от дистрибутива и версии ядра.

Лимиты System V IPC

Параметры ядра для System V очередей:

  • kernel.msgmax - максимальный размер одного сообщения в байтах, по умолчанию 8192
  • kernel.msgmnb - максимальный суммарный размер всех сообщений в одной очереди, по умолчанию 16384
  • kernel.msgmni - максимальное количество очередей в системе, по умолчанию 32000

Просмотр текущих значений:

sysctl kernel.msgmax kernel.msgmnb kernel.msgmni
ipcs -l

Изменение на лету:

sysctl -w kernel.msgmax=16384
sysctl -w kernel.msgmnb=32768

Для сохранения после перезагрузки добавить строки в /etc/sysctl.conf или файл в /etc/sysctl.d/.

Лимиты POSIX mqueue

Параметры POSIX очередей находятся в /proc/sys/fs/mqueue/:

  • msg_max - максимальное количество сообщений в одной очереди, по умолчанию 10
  • msgsize_max - максимальный размер одного сообщения, по умолчанию 8192
  • queues_max - максимальное количество очередей в системе, по умолчанию 256

Просмотр:

cat /proc/sys/fs/mqueue/msg_max
cat /proc/sys/fs/mqueue/msgsize_max
cat /proc/sys/fs/mqueue/queues_max

Изменение:

sysctl -w fs.mqueue.msg_max=100
sysctl -w fs.mqueue.msgsize_max=16384
sysctl -w fs.mqueue.queues_max=1024

Для постоянного применения добавить в /etc/sysctl.conf. Увеличение лимитов повышает потребление памяти ядром, поэтому менять значения нужно осознанно, под фактическую нагрузку.

Типичные ошибки и подводные камни

Разработчики и администраторы сталкиваются с одними и теми же проблемами при работе с очередями сообщений. Разбор каждой ошибки включает симптом, причину и решение.

Обработка неблокирующих операций

Симптом: программа периодически завершается с ошибкой «Resource temporarily unavailable» или «Try again». Причина: при установленном IPC_NOWAIT или O_NONBLOCK функции возвращают -1 с errno = EAGAIN, когда операция не может быть выполнена немедленно. Код не проверяет errno и трактует любую ошибку как фатальную.

Решение: обрабатывать EAGAIN отдельно. Для отправки - повторить попытку позже или залогировать переполнение очереди. Для получения - выйти из цикла обработки или перейти к другим задачам.

#include 
#include 
#include 

int main() {
    mqd_t mq = mq_open("/myqueue", O_RDONLY | O_NONBLOCK);
    if (mq == (mqd_t)-1) {
        perror("mq_open");
        return 1;
    }

    char buffer[8192];
    ssize_t n = mq_receive(mq, buffer, sizeof(buffer), NULL);
    if (n == -1) {
        if (errno == EAGAIN) {
            printf("Очередь пуста, продолжаем работу\n");
        } else {
            perror("mq_receive");
        }
    } else {
        printf("Получено: %s\n", buffer);
    }
    mq_close(mq);
    return 0;
}

Утечки ресурсов и очистка очередей

Симптом: после многократных запусков программы новые очереди перестают создаваться, в логах ошибка ENOSPC. Причина: очереди System V существуют до вызова msgctl с IPC_RMID или перезагрузки. Очереди POSIX - до вызова mq_unlink. Каждая забытая очередь занимает память ядра и расходует лимит на количество очередей.

Решение: всегда удалять очереди при завершении работы. Для System V - вызывать msgctl(msgid, IPC_RMID, NULL) в обработчике завершения. Для POSIX - вызывать mq_close и mq_unlink. Просмотр существующих очередей System V: команда ipcs -q. Удаление вручную: ipcrm -q <msqid>. Для POSIX очередей просмотр доступен через смонтированную mqueuefs: mount -t mqueue mqueue /dev/mqueue, затем ls /dev/mqueue.

Когда использовать очереди сообщений: сравнение с другими IPC

Очереди сообщений подходят для обмена дискретными сообщениями с сохранением границ и приоритетов. Pipe передаёт поток байтов без границ: получатель не знает, где заканчивается одно сообщение и начинается другое, если формат не задан вручную. Socket добавляет сетевые накладные расходы и сложность, но позволяет общаться через сеть. Shared memory даёт максимальную скорость, но требует отдельной синхронизации через семафоры или мьютексы.

Сценарии, где очереди сообщений удобны: передача команд между процессами демона и клиента, буферизация логов, обмен данными между рабочими процессами, реализация пула задач с приоритетами. Если сообщения должны обрабатываться асинхронно несколькими потребителями или нужна гарантированная доставка при сбоях, стоит рассмотреть брокер. Статья об архитектуре брокеров сообщений разбирает durable-очереди, TTL и Dead Letter Exchange в RabbitMQ.

Для интеграции разнородных систем через сеть, включая 1C и legacy-приложения, практическое руководство по настройке RabbitMQ показывает паттерны гарантированной доставки и мониторинга.

Заключение

Очереди сообщений в Linux решают задачу межпроцессного обмена дискретными данными без внешних сервисов. System V IPC - проверенный механизм с функциями msgget, msgsnd, msgrcv, msgctl. POSIX mqueue - современный стандарт с mq_open, mq_send, mq_receive, mq_close, mq_unlink и асинхронными уведомлениями через mq_notify.

Для новых проектов выбирайте POSIX mqueue: интерфейс проще, переносимость выше, есть уведомления. Для поддержки существующего кода на System V сохраняйте текущий механизм. Следите за лимитами ядра, удаляйте очереди после использования и обрабатывайте EAGAIN в неблокирующем режиме. Практикуйтесь на примерах из этой статьи и проверяйте поведение на своей версии ядра.

Поделиться:
Сохранить гайд? В закладки браузера