> For the complete documentation index, see [llms.txt](https://kopens.gitbook.io/plantpulse-platform/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://kopens.gitbook.io/plantpulse-platform/admin/modules/messaging.md).

# messaging (메시징)

## 역할

PlantPulse 의 **신경계** 역할을 하는 메시지 브로커 계층입니다. 산업 현장에서 들어오는 센서 데이터를 수집하고, 모듈 간 이벤트를 전달하며, 콘솔로 실시간 푸시를 전달합니다.

| 항목     | 값                                                                        |
| ------ | ------------------------------------------------------------------------ |
| 모듈명    | `plantpulse-messaging`                                                   |
| 설치 경로  | `/opt/kopens/plantpulse-platform/plantpulse-messaging/`                  |
| 데이터 위치 | `/data1/pp-data/kafka/`, `/data1/pp-data/mqtt/`, `/data1/pp-data/stomp/` |

## 구성

```mermaid
graph LR
  subgraph EDGE["산업 현장"]
    SENSOR[IoT 센서]
    PLC[PLC / 제어기]
    AGENT[Edge Agent]
  end

  subgraph MSG["plantpulse-messaging"]
    MQTT[MQTT (HiveMQ)<br/>:1883 / :1884 TLS]
    KAFKA[Kafka<br/>:9092 / :9093 controller / :9094 TLS]
    STOMP[STOMP (ActiveMQ)<br/>:61000 / :61004 TLS]
  end

  subgraph CONSUMER["소비자"]
    SVR[plantpulse-server]
    CEP[plantpulse-cep]
    BAT[plantpulse-batch]
    WS[WebSocket 푸시]
    BROWSER[브라우저]
  end

  SENSOR --> MQTT
  PLC --> AGENT
  AGENT --> KAFKA
  MQTT --> KAFKA
  KAFKA --> SVR
  KAFKA --> CEP
  KAFKA --> BAT
  SVR --> STOMP
  STOMP --> WS
  WS --> BROWSER
```

## 각 브로커 상세

### Kafka — 대용량 데이터 스트리밍

산업 데이터의 핵심 메시지 버스. KRaft 모드 (Zookeeper 없음).

| 항목     | 값                                               |
| ------ | ----------------------------------------------- |
| 버전     | 3.7+ (KRaft 모드)                                 |
| 포트     | 9092 (PLAINTEXT), 9093 (controller), 9094 (TLS) |
| 인증     | SASL/PLAIN (사용자 `mq`, `PP_MQ_USER`)             |
| 토픽 접두사 | `pp` (`PP_TOPIC_PREFIX`)                        |

#### 주요 토픽

| 토픽                        | 용도                   |
| ------------------------- | -------------------- |
| `pp-tag-point`            | 태그 값 변경 이벤트 (메인 스트림) |
| `pp-tag-alarm`            | 태그 알람 이벤트            |
| `pp-asset-data`           | 에셋 집계 데이터            |
| `pp-asset-alarm`          | 에셋 알람 이벤트            |
| `pp-asset-event`          | 에셋 이벤트               |
| `pp-domain-changed-event` | 메타데이터 변경 이벤트         |
| `pp-cep-output`           | CEP 처리 결과            |
| `pp-flow-trigger`         | Flow 엔진 트리거          |

#### 디렉토리 구조

```
plantpulse-messaging/kafka/
├── bin/
│   ├── kafka-server-start.sh
│   ├── kafka-server-stop.sh
│   ├── kafka-topics.sh
│   ├── kafka-console-consumer.sh
│   ├── kafka-console-producer.sh
│   └── kafka-consumer-groups.sh
├── config/
│   ├── broker.properties            # 핵심 설정
│   ├── server.properties
│   ├── controller.properties
│   ├── consumer.properties
│   ├── producer.properties
│   ├── connect-distributed.properties
│   ├── connect-mirror-maker.properties
│   └── log4j2.yaml
├── libs/
└── logs/
    ├── server.log                    # 메인 로그
    └── controller.log
```

#### 핵심 설정 (`config/broker.properties`)

```properties
node.id=1
process.roles=broker,controller
controller.quorum.voters=1@${PP_KAFKA_HOST}:9093

listeners=PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093,SSL://0.0.0.0:9094
advertised.listeners=PLAINTEXT://${PP_PUBLIC_IP}:9092,SSL://${PP_PUBLIC_IP}:9094

log.dirs=/data1/pp-data/kafka

# 토픽 기본
num.partitions=12
default.replication.factor=1     # 단일 노드. 클러스터에서는 3
min.insync.replicas=1

# 보존
log.retention.hours=168           # 7일
log.retention.bytes=-1            # 크기 무제한
log.segment.bytes=1073741824      # 1GB

# 성능
num.network.threads=8
num.io.threads=16
socket.send.buffer.bytes=1048576
socket.receive.buffer.bytes=1048576

# 보안 (SASL/PLAIN)
sasl.enabled.mechanisms=PLAIN
listener.security.protocol.map=PLAINTEXT:SASL_PLAINTEXT,CONTROLLER:SASL_PLAINTEXT,SSL:SASL_SSL
```

#### 운영 명령

```bash
cd /opt/kopens/plantpulse-platform/plantpulse-messaging/kafka/bin

# 토픽 목록
./kafka-topics.sh --bootstrap-server localhost:9092 --list

# 토픽 생성
./kafka-topics.sh --bootstrap-server localhost:9092 --create \
  --topic pp-test --partitions 12 --replication-factor 1

# 토픽 상세
./kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic pp-tag-point

# Consumer Group 상태 (lag 확인)
./kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list
./kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group server-consumer

# 메시지 직접 보기 (디버그)
./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic pp-tag-point --from-beginning --max-messages 10
```

`plantpulse-startup/node-topic.sh` 가 위 명령을 wrapping 하여 비밀번호 자동 주입까지 처리합니다.

```bash
/opt/kopens/plantpulse-platform/plantpulse-startup/node-topic.sh
```

#### Kafka 튜닝 포인트

| 항목                            | 위치                          | 권장                      |
| ----------------------------- | --------------------------- | ----------------------- |
| 힙                             | `bin/kafka-server-start.sh` | 4\~8GB (페이지 캐시 OS 에 위임) |
| 파티션 수                         | 토픽별                         | 코어 수의 1\~2배             |
| `num.io.threads`              | broker.properties           | 16\~32 (NVMe)           |
| `log.flush.interval.messages` | broker.properties           | 기본값 유지 (OS sync)        |

### MQTT (HiveMQ) — IoT 디바이스 수집

경량 IoT 디바이스 / 센서로부터 데이터를 수집하는 표준 IoT 프로토콜.

| 항목    | 값                                |
| ----- | -------------------------------- |
| 브로커   | HiveMQ Community Edition         |
| 포트    | 1883 (PLAINTEXT), 1884 (TLS)     |
| 외부 포트 | 18883 / 18884 (대외 노출용)           |
| 인증    | username/password (`PP_MQ_USER`) |

#### 디렉토리 구조

```
plantpulse-messaging/mqtt/
├── bin/run.sh
├── conf/
│   ├── config.xml                # 메인 설정
│   ├── auth.properties           # 인증 / ACL
│   └── logback.xml
├── extensions/                    # HiveMQ 확장 플러그인
├── diagnostics/
└── logs/
    └── hivemq.log
```

#### 핵심 설정 (`conf/config.xml`)

```xml
<hivemq>
  <listeners>
    <tcp-listener>
      <port>1883</port>
      <bind-address>0.0.0.0</bind-address>
    </tcp-listener>
    <tls-tcp-listener>
      <port>1884</port>
      <tls>
        <keystore>
          <path>/var/security/plantpulse/mqtt.jks</path>
          <password>${PP_TLS_KEYSTORE_PASSWORD}</password>
        </keystore>
      </tls>
    </tls-tcp-listener>
  </listeners>
  <mqtt>
    <session-expiry>
      <max-interval>86400</max-interval>
    </session-expiry>
    <message-expiry>
      <max-interval>3600</max-interval>
    </message-expiry>
    <queued-messages>
      <max-queue-size>1000</max-queue-size>
    </queued-messages>
  </mqtt>
</hivemq>
```

#### 운영 / 디버깅

```bash
# 외부에서 토픽 발행 (디버깅)
mosquitto_pub -h 127.0.0.1 -p 1883 -u mq -P mq123! -t test/topic -m "hello"

# 구독 확인
mosquitto_sub -h 127.0.0.1 -p 1883 -u mq -P mq123! -t test/#
```

### STOMP (ActiveMQ) — 콘솔 실시간 푸시

WebSocket 기반으로 콘솔 브라우저에 이벤트를 실시간으로 푸시합니다.

| 항목      | 값                              |
| ------- | ------------------------------ |
| 브로커     | Apache ActiveMQ                |
| 포트      | 61000 (PLAINTEXT), 61004 (TLS) |
| 콘솔 (관리) | 8161                           |

#### 디렉토리 구조

```
plantpulse-messaging/stomp/
├── bin/activemq
├── conf/
│   ├── activemq.xml              # 브로커 설정
│   ├── jetty.xml
│   └── jetty-realm.properties     # 인증
├── data/                          # KahaDB
├── webapps/                       # 콘솔 UI
└── logs/activemq.log
```

#### 핵심 토픽 / 큐

| 큐                     | 용도         |
| --------------------- | ---------- |
| `/topic/alarm`        | 알람 푸시      |
| `/topic/data`         | 실시간 데이터 푸시 |
| `/topic/event`        | 시스템 이벤트    |
| `/queue/notification` | 사용자 알림     |

## 통합 운영 명령

```bash
cd /opt/kopens/plantpulse-platform/plantpulse-startup

./restart-messaging.sh    # 메시징 모듈 전체 재시작 (Kafka + MQTT + STOMP)
./node-topic.sh           # Kafka 토픽 관리 대화형 메뉴
./status.sh               # 9092 / 1883 / 61000 RUNNING 확인
```

> **재시작 영향**: 메시징 재시작 시 모든 메시지 수신이 일시 중단됩니다 (수초). 의존 모듈(server, cep, batch 등) 도 함께 재시작하시는 것이 안전합니다.

## 로그 위치

| 브로커   | 메인 로그                                          |
| ----- | ---------------------------------------------- |
| Kafka | `plantpulse-messaging/kafka/logs/server.log`   |
| MQTT  | `plantpulse-messaging/mqtt/logs/hivemq.log`    |
| STOMP | `plantpulse-messaging/stomp/logs/activemq.log` |

통합 로그 뷰어로 한 번에 보기:

```bash
/opt/kopens/plantpulse-platform/plantpulse-startup/log-viewer.sh
```

## 자주 발생하는 문제

| 증상                          | 원인                   | 조치                                                         |
| --------------------------- | -------------------- | ---------------------------------------------------------- |
| Kafka 메시지 적체 (consumer lag) | 소비 처리량 부족            | `kafka-consumer-groups.sh --describe` 로 lag 확인, 컨슈머 스레드 증가 |
| Kafka 디스크 가득                | retention 미설정 / 큰 토픽 | `log.retention.hours` 조정, 불필요 토픽 삭제                        |
| MQTT 연결 폭주                  | session expiry 누적    | `session-expiry.max-interval` 단축                           |
| MQTT TLS handshake 실패       | 인증서 만료 / SAN 누락      | `prepare-ssl.sh` 재실행, `PP_TLS_SAN_IPS` 점검                  |
| STOMP 콘솔 메시지 안 옴            | WebSocket 포트 방화벽     | 8000/8004 포트 확인                                            |
| Kafka controller 선출 실패      | KRaft 메타 손상          | `data/kafka/__cluster_metadata-0` 백업 후 운영팀 문의              |

## 보안 / 외부 노출

| 포트                      | 보호 권장      |
| ----------------------- | ---------- |
| 9092 (Kafka PLAINTEXT)  | 사설망 only   |
| 9094 (Kafka TLS)        | 외부 노출 시 사용 |
| 1883 (MQTT)             | 사설망 only   |
| 1884 (MQTT TLS)         | 외부 노출 시 사용 |
| 18883 / 18884 (MQTT 외부) | 외부 노출 가능   |
| 61000 (STOMP)           | 사설망 only   |
| 61004 (STOMP TLS)       | 외부 노출 시 사용 |

운영 환경에서는 TLS 포트만 외부 노출하고 PLAINTEXT 포트는 사설망 / Tailscale 등으로 제한해 주세요.

## 클러스터 / HA

단일 노드 운영에서 다음과 같이 확장 가능합니다.

| 브로커   | HA 방식                                                 |
| ----- | ----------------------------------------------------- |
| Kafka | 브로커 3대 + replication-factor 3 + min.insync.replicas 2 |
| MQTT  | HiveMQ Enterprise (`mqtt-ee/`) 클러스터                   |
| STOMP | ActiveMQ Master-Slave (KahaDB 공유)                     |

상세 클러스터 설계는 [클러스터 설치](/plantpulse-platform/installation/build-deploy.md) 페이지를 참고해 주세요.

## 관련 문서

* [모듈 인덱스](/plantpulse-platform/admin/modules.md)
* [메시징 아키텍처 (개발자)](/plantpulse-platform/developer/messaging.md)
* [데이터 처리 플로우](/plantpulse-platform/developer/data-flow.md)
* [성능 튜닝](/plantpulse-platform/admin/performance-tuning.md)
* [문제 해결](/plantpulse-platform/admin/troubleshooting.md)