Python socket の使い方｜TCP/IP で産業機器とリアルタイム通信する実戦ガイド

import socket

with socket.create_connection(("127.0.0.1", 9000), timeout=5.0) as sock:  # 機器の IP に置き換える
    sock.settimeout(5.0)          # 接続後の recv / sendall に効くタイムアウト
    sock.sendall(b"PING\n")
    print(sock.recv(1024))        # 「最大 1024 バイト」であって「1024 バイト」ではない

def recv_exact(sock, n):
    buf = bytearray()
    while len(buf) < n:
        chunk = sock.recv(n - len(buf))
        if not chunk:                        # 0 バイト = 相手が閉じた
            raise ConnectionError(f"peer closed: got {len(buf)} of {n} bytes")
        buf.extend(chunk)
    return bytes(buf)

import struct

HEADER = struct.Struct(">I")                  # 4 バイト big-endian。外部通信では @ を使わない

def recv_frame(sock):
    length, = HEADER.unpack(recv_exact(sock, HEADER.size))
    return recv_exact(sock, length)

sock.settimeout(5.0)
try:
    frame = recv_frame(sock)
except TimeoutError:                          # socket.timeout は現在この別名
    ...

listener = socket.socket()
if hasattr(socket, "SO_EXCLUSIVEADDRUSE"):    # Windows でのみ定義される
    listener.setsockopt(socket.SOL_SOCKET, socket.SO_EXCLUSIVEADDRUSE, 1)
listener.bind(("127.0.0.1", 9000))
listener.listen()

2026 年 8 月 31 日更新: Windows 11 Pro / Python 3.12.10 で全体を再検証し、「socket の使い方の全体像」「API 早見表」「struct でのバイナリ電文」「タイムアウトの 3 モード」「よくあるエラーの実文言」を新設しました。あわせて、SO_REUSEADDR は Windows では意味がまるで違う（自分のサーバーを他プロセスに乗っ取らせる設定になる）ことを実測で確認し、旧版のサンプルコードから削除しています。socket.timeout の位置づけ（現在は TimeoutError の非推奨エイリアス）と、TCP_KEEPCNT の対応バージョンも一次情報に合わせて訂正しました。

1. どんなときに socket を自前で書くのか

産業機器との通信は、Modbus TCP や OPC UA のような標準プロトコルが使えれば話が早く、専用ライブラリに任せられます。それでも現場には「メーカー独自のバイナリプロトコル」「仕様書（PDF）のとおりにバイト列を組み立てて送る」という機器が残ります。温度センサ、カウンタ、独自プロトコルの計測器——汎用プロトコルを実装していない機器は珍しくありません。こうした相手には、Python 標準の socket モジュールで自前実装するのが最も汎用性の高い解になります。

先に分岐を示します。標準プロトコルで足りるなら、この記事の内容は書かなくて済みます。

• 機器が Modbus TCP に対応している → pymodbus の記事（フレーム組み立ても再送も済んでいる）
• 三菱・キーエンス・オムロン・シーメンスの PLC → PLC 通信の記事（メーカー別ライブラリの比較）
• 相手が RS-232C / RS-485 のシリアル機器 → pyserial の記事（物理層が違うだけで、フレーム境界の悩みはよく似ています）
• 上のどれでもない独自プロトコル → 本記事

なお本記事は TCP のみを扱います。UDP（SOCK_DGRAM）は「届いたかどうか・順番・重複」をアプリ側で面倒みる前提のプロトコルで、装置の制御コマンドやログ収集のように1 件も落としたくない用途では TCP を選ぶのが基本です。逆に、多少落ちてもよい高頻度のモニタ値配信や、1 対多のブロードキャストでは UDP が候補になります。

2. 使い方の全体像 — socket の 6 ステップ

TCP 通信は「待ち受ける側（サーバー）」と「つなぎに行く側（クライアント）」の役割分担で動きます。呼ぶ関数は 6 つだけです。

最初の関門は 4 と 5 の関係です。accept() が返すソケットと、listen() しているソケットは別物で、データのやり取りは前者で行います。待ち受け用ソケットは「接続を受け付ける係」であり、そこに recv() しても機器のデータは届きません。

2.1 どちらが bind するかは仕様書で決まる

機器が「常時待ち受けるサーバー」なら Python 側がクライアントとして接続しに行きます。機器側がイベント発生時に通知を送ってくる構成なら、Python 側がサーバーとして待ち受けます。仕様書を読むときは、まずどちらが bind するのかを確認してください。ここを取り違えると、コードの半分が無駄になります。

Python 側がサーバーとして待ち受ける構成を会社 PC でやる場合、Windows ファイアウォールの受信許可が必要です。管理者権限のない PC では自分で許可を追加できないため、情報システム部門への依頼を見込んでおいてください。「ローカルで動いたのに現場 PC でだけ接続できない」は、たいていこの受信許可か、機器側の接続元 IP 制限が原因です。

2.2 最小実装 — ダミーサーバーとクライアント

実機が手元に無くても、サーバー役のスクリプトを 1 つ用意すれば PC 単体で動作確認が完結します。まずは受け取った電文に ACK を返すだけのダミーサーバーです。

# server_simulator.py — 自宅 PC で動かす機器シミュレータ（受け取った電文に ACK を返すだけ）
import socket

HOST, PORT = "127.0.0.1", 9000

listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
if hasattr(socket, "SO_EXCLUSIVEADDRUSE"):   # Windows でのみ定義される（9 章）
    listener.setsockopt(socket.SOL_SOCKET, socket.SO_EXCLUSIVEADDRUSE, 1)

with listener:
    listener.bind((HOST, PORT))
    listener.listen()
    print(f"listening on {HOST}:{PORT}")
    while True:                       # 1 接続で終わらせず、切れたらまた待ち受ける
        conn, addr = listener.accept()
        with conn:
            print(f"connected: {addr}")
            while True:
                data = conn.recv(1024)
                if not data:          # 0 バイト = 相手が接続を閉じた
                    break
                print(f"recv: {data!r}")
                conn.sendall(b"ACK\n")
        print("disconnected")

# client.py — Python から機器に接続する側
import socket

HOST, PORT = "127.0.0.1", 9000

with socket.create_connection((HOST, PORT), timeout=5.0) as sock:
    sock.settimeout(5.0)              # 接続後の recv / send に効くタイムアウト
    sock.sendall(b"PING\n")
    response = sock.recv(1024)
    print(f"response: {response!r}")
    sock.shutdown(socket.SHUT_WR)     # 「もう送らない」を相手に伝えてから閉じる

動かし方は、PowerShell（またはコマンドプロンプト）を 2 つ開き、1 つ目で python server_simulator.py（listening on 127.0.0.1:9000 と出て待機します）、2 つ目で python client.py を実行します。クライアント側に response: b'ACK\n' と表示されれば成功です（当サイトの実行結果）。

旧版のサンプルから変えた点が 3 つあります。①SO_REUSEADDR を外しました（Windows では自分のサーバーを乗っ取り可能にする設定です。9 章）。②accept() をループにしました（1 接続で終わるサーバーは、クライアントを試すたびに再起動が要ります）。③クライアントに create_connection とタイムアウトを使いました（connect は放置すると OS 任せの時間だけ待ちます。6 章）。

2.3 このコードが現場で壊れる 4 つの理由

ここまでで「動く」状態にはなりました。ただし上のクライアントは、次の 4 つを暗黙に仮定しています。現場ではすべて成り立ちません。

• 1 回の sendall が 1 回の recv で届くと仮定している → TCP に境界はありません（4 章）
• recv(1024) が要求どおり返ると仮定している → 1 バイトしか返らないこともあります（4 章）
• 相手が必ず応答すると仮定している → 黙られたときの振る舞いはタイムアウト設定次第です（6 章）
• 切れたら例外が出ると仮定している → ケーブルが抜けても TCP は黙ったままのことがあります（8 章）

3. socket API 早見表

仕様書と首っ引きで書いているときに引く用の表です。戻り値と例外の欄が、現場でのハマりどころに直結します。

sendall の注意は公式ドキュメントの明記事項です。“On error, an exception is raised, and there is no way to determine how much data, if any, was successfully sent.”（エラー時に、どれだけ送れたのかを知る方法は無い）。送信が途中で失敗したとき、機器が「半分だけ受け取った」状態になり得るということです。だからコマンド送信は、再送しても害の無い設計（同じコマンドを 2 回受けても問題ない、あるいは通し番号で重複を弾ける）にしておくのが安全です。

4. recv はストリーム — 受信データが途中で切れる理由とフレーム境界の作り方

TCP はバイトストリームです。送った単位も、区切りも、受信側には伝わりません。公式のソケットプログラミング HOWTO は「メッセージが完全に処理されるまで繰り返し呼び出すのは 自分の 責任なのだ」と書いています。recv を 1 回呼んで終わりにしてよい場面は、実運用ではほぼありません。

4.1 実測 — くっつく方向にも、割れる方向にも壊れる

当サイトの検証環境（Windows 11 Pro / Python 3.12.10 / 127.0.0.1）で 2 つ確かめました。

• 結合: 送信側が sendall(b"A"*10) sendall(b"B"*10) sendall(b"C"*10) と 3 回に分けて送り、受信側が少し待ってから recv(1024) を 1 回呼ぶと、30 バイトがまとめて返りました（b'AAAAAAAAAABBBBBBBBBBCCCCCCCCCC'）
• 分割: 送信側が 102,400 バイトを 1 回の sendall で送ると、受信側の recv(65536) は 65,536 バイト → 36,864 バイトの 2 回に分かれました。recv(4096) にすると 25 回です

ループバック（同一 PC 内の折り返し）でもこの通りで、実ネットワークでは経路の MTU や輻輳（ネットワークが混雑した状態）でさらに細かく割れます。「小さいデータなら 1 回で届くから大丈夫」という経験則は、負荷が上がった日に壊れます。

4.2 境界の付け方は 3 通り、常駐用途で選べるのは 2 つ

タイマ方式は、シリアル通信（Modbus RTU の 3.5 文字時間など）では成立します。TCP で成立しない理由は、時間軸が信用できないからです。送信側の Nagle アルゴリズムが小さなデータをまとめて遅らせ（10 章）、受信側の遅延 ACK（受信確認をわずかに遅らせる TCP の既定動作）がさらに間を空け、経路の途中でも再分割・再送が起きます。「20 ミリ秒空いたから 1 フレーム終わり」という判定は、ネットワークが少し混んだ日に誤検知します。同じ境界問題でもシリアルではタイマ方式が使え、TCP では使えない——この差は pyserial 記事のパケット境界の章と読み比べると分かりやすいはずです。

4.3 recv_exact と recv_frame

# framing.py — 長さプレフィックス方式の送受信ヘルパー
import socket
import struct

HEADER = struct.Struct(">I")          # 4 バイトの符号なし整数（big-endian）
MAX_FRAME = 1024 * 1024               # 想定外の長さでメモリを食い潰さない上限


def recv_exact(sock: socket.socket, n: int) -> bytes:
    """正確に n バイト受信するまで繰り返し読む。相手が閉じたら例外にする。"""
    buf = bytearray()
    while len(buf) < n:
        chunk = sock.recv(n - len(buf))
        if not chunk:                 # 0 バイト = 相手が接続を閉じた
            raise ConnectionError(f"peer closed: got {len(buf)} of {n} bytes")
        buf.extend(chunk)
    return bytes(buf)


def recv_frame(sock: socket.socket) -> bytes:
    """長さプレフィックス（4 バイト big-endian）方式で 1 フレーム受信する。"""
    length, = HEADER.unpack(recv_exact(sock, HEADER.size))
    if length > MAX_FRAME:
        raise ConnectionError(f"frame too large: {length}")
    return recv_exact(sock, length)


def send_frame(sock: socket.socket, payload: bytes) -> None:
    """長さプレフィックスを付けて 1 フレーム送信する。"""
    sock.sendall(HEADER.pack(len(payload)) + payload)

ポイントは 3 つです。①受信箇所をすべて recv_frame に一本化する（生の recv がコードに残っていると、そこだけ境界問題が再発します）。②長さの上限を必ず設ける（ノイズやバージョン違いで長さが化けたとき、recv_exact(4294967295) を実行しないため）。③相手が閉じた場合を例外に変換する（b"" のまま処理を続けると、無限ループか無音故障になります）。

なお、TCP ではアプリ層の CRC は原則不要です。TCP 自身がセグメントごとのチェックサムと再送で誤りを訂正するため、化けたデータはそのまま上がってきません。シリアル通信で CRC が必須なのは、物理層にその仕組みが無いからです（pyserial 記事の CRC の章）。機器の仕様書が CRC を要求している場合は、それは相手プロトコルの仕様として実装します。

5. struct でバイナリ電文を組む — バイト数が合わない事故の正体

仕様書どおりにフォーマット文字列を書いたのに、パースするとバイト数が合わない。struct で最も多い事故は、先頭に付けるバイトオーダー指定子の取り違えです。

指定子を書かないと @ になり、C コンパイラと同じ規則で詰め物（パディング）が自動的に入ります。公式ドキュメントもそのものずばりを書いています。“When exchanging data beyond your process such as networking or storage, be precise. Specify the exact byte order, size, and alignment.”（プロセスの外——ネットワークや保存——とデータをやり取りするなら、バイト順・サイズ・アライメントを厳密に指定せよ）。当サイトの実測でも、同じ並びで数値が変わります。

仕様書には「9 バイトの電文」と書いてあるのに、struct.calcsize が 12 を返す。「バイト数が合わない」の正体はこれです。外部と交換する電文では、先頭に <・>・! のいずれかを必ず書いてください。詰め物は自分で管理する（＝仕様書に予約バイトがあるなら x で明示的に書く）のが公式の考え方です。

import struct

# 温度センサのダミー電文: 種別 1B / 通し番号 2B / 温度 x100 の符号付き 2B / 経過秒 4B
TELEGRAM = struct.Struct("<BHhI")      # Struct を 1 度作って使い回すと解析が 1 回で済む

raw = TELEGRAM.pack(0x01, 1234, -1250, 86400)
print(raw.hex(" "), len(raw))          # 01 d2 04 1e fb 80 51 01 00 9

kind, seq, temp_x100, uptime = TELEGRAM.unpack(raw)
print(f"kind={kind} seq={seq} temp={temp_x100 / 100:.2f} C uptime={uptime}s")
# kind=1 seq=1234 temp=-12.50 C uptime=86400s

try:
    TELEGRAM.unpack(raw[:-1])          # 1 バイト足りない
except struct.error as e:
    print("struct.error:", e)          # unpack requires a buffer of 9 bytes

実装のコツを 3 つ。①温度や電流は「x100 の整数」で送られてくることが多いので、受信直後に / 100 して実単位に直す層を用意しておくと、以降の処理が読みやすくなります。②struct.Struct をモジュール定数にする——公式も「フォーマット文字列の解析が 1 回で済むぶん、モジュール関数を呼ぶより効率的」と書いています。③長さ不足は struct.error になるので、recv_exact で必要バイト数を揃えてから unpack する順序を守ってください。struct の例外は OSError の仲間ではないため、通信の例外処理とは別に捕まえる必要があります。

6. タイムアウトの 3 モード

ソケットには 3 つのモードがあります（公式ドキュメントの “blocking, non-blocking, or timeout”）。切り替えるのは settimeout() ひとつで、渡す値によって読み取りの性格そのものが変わります。

図 2 の 4 レーン目が、設計を誤りやすい点です。タイムアウトは「フレームが揃うまでの制限時間」ではなく「次の 1 バイトが届くまでの制限時間」です。settimeout(3.0) で recv(1024) を呼び、0.05 秒後に 2 バイトだけ届いた実測では、recv は 0.051 秒で 2 バイトを返しました。だから recv_exact のループが必要になります。

6.1 捕まえる例外は TimeoutError（socket.timeout は別名）

捕まえる例外の名前も更新が必要です。socket.timeout は公式ドキュメントで “A deprecated alias of TimeoutError.”（TimeoutError の非推奨エイリアス）とされています。当サイトの環境でも socket.timeout is TimeoutError は True でした。新しく書くコードでは TimeoutError を捕まえてください。TimeoutError は OSError の派生なので、except OSError でまとめて拾うと通信エラー全般と同じ扱いにできます。

# ※断片です（HOST / PORT / recv_frame / logger は定義済みの前提。動く形は 11 章の DeviceClient）
# タイムアウトの設定は「接続用」と「接続後」で分ける
sock = socket.create_connection((HOST, PORT), timeout=5.0)  # 接続にかける制限時間
sock.settimeout(5.0)                                        # 接続後の recv / sendall の制限時間

try:
    frame = recv_frame(sock)
except TimeoutError:            # socket.timeout は現在この別名（args は ('timed out',)）
    logger.warning("機器から 5 秒間フレームが来ない")
    raise                       # 再接続フローへ流す

connect のタイムアウトには注意があります。公式は「connect() の前に settimeout() を呼ぶか、create_connection() に timeout を渡すことを推奨する。ただし OS のネットワークスタックが、Python 側の設定と関係なく独自の接続タイムアウトエラーを返すこともある」と書いています。「30 秒に設定したのに 20 秒で例外が出た」は仕様の範囲内です。

もう 1 点、sendall のタイムアウトは Python 3.5 以降「全データを送り終えるまでの合計時間」です（送信が進むたびにリセットされる仕様ではありません）。大きなデータを送る場合、タイムアウト値は本文サイズを考慮して決めてください。

7. つながらない・切れたを見分ける（エラー早見表）

socket のエラーは OSError の派生で、Windows では WinError 番号が付きます。番号が分かれば原因はほぼ絞れます。下表のメッセージは、当サイトの環境（Windows 11 日本語版 / Python 3.12.10）で実際に採取したものです。

接続失敗にかかる時間も、実測すると想像とずれます。当サイトの環境で、誰も待ち受けていない 127.0.0.1 のポートへ接続すると、ConnectionRefusedError が返るまで約 2.0 秒かかりました（3 回とも 2.03〜2.07 秒）。即座にエラーになるわけではありません。さらに ホスト名 localhost を指定すると約 4.1 秒に伸びました。create_connection はホスト名を IPv6 と IPv4 の両方に解決し、成功するまで順に試すためです（約 2 秒 × 2 回）。再接続の間隔を 1 秒に設定していても、1 回の試行に 2〜4 秒かかる前提で設計してください。

「どのアドレスで失敗したのか分からない」を解決するのが all_errors=True（Python 3.11 以降）です。既定では最後に試したアドレスの例外しか上がりませんが、True にすると全部の失敗が ExceptionGroup にまとまります。

# ※断片です（socket / logger は import・定義済みの前提。動く形は 11 章の device_client.py）
try:
    sock = socket.create_connection(("device.local", 9000), timeout=3, all_errors=True)
except* OSError as eg:                     # except* は ExceptionGroup 用（Python 3.11 以降）
    for e in eg.exceptions:
        logger.error("接続失敗: %s", e)
# 実測（localhost の空きポート）:
#   ExceptionGroup: create_connection failed (2 sub-exceptions)
#     - ConnectionRefusedError [WinError 10061] ...   ← IPv6 側
#     - ConnectionRefusedError [WinError 10061] ...   ← IPv4 側

8. 切断検知と TCP Keep-Alive

切断には 2 種類あります。相手がきちんと閉じた場合は、recv が b""（0 バイト）を返します。公式 HOWTO の表現を借りれば「recv が 0 バイトを返したときは、向こう側が接続を閉じてしまった (または閉じようとしている途中) という意味だ。もうこの接続でデータを受け取ることはない。永遠にだ」。当サイトでも、送信側が shutdown(SHUT_WR) した直後に受信側の recv が b'' を返すことを確認しています。

問題はもう 1 種類のほうです。LAN ケーブルが抜けた、ハブの電源が落ちた、機器が固まった。この場合、TCP は何も言いません。こちらが送信を試みるまで異常に気づけず、受信専用のコードは永久に静かなままになり得ます。ここで出てくるのが TCP Keep-Alive ですが、既定値がそのまま使えません。

8.1 Keep-Alive の既定値は OS で異なる

有効にしても、既定ではアイドル 2 時間後にようやく検知が始まります。24 時間動く監視アプリで「2 時間気づかない」は使いものになりません。当サイトの Windows 11 環境で getsockopt を読むと、確かに TCP_KEEPIDLE = 7200（秒）、TCP_KEEPINTVL = 1、TCP_KEEPCNT = 10 が既定で入っていました（Microsoft のドキュメントの記述と一致します）。値は次のように詰められます。

# ※関数の断片です（socket を import している前提。11 章の device_client.py に組み込み済み）
def enable_tcp_keepalive(sock: socket.socket, idle: int = 30,
                         interval: int = 5, count: int = 4) -> None:
    """OS の TCP Keep-Alive を有効にし、間隔を詰める（定数が無い OS では黙って飛ばす）。"""
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
    for name, value in (("TCP_KEEPIDLE", idle),        # Python 3.7+ / Windows 10 1709 以降
                        ("TCP_KEEPINTVL", interval),   # Python 3.7+ / Windows 10 1709 以降
                        ("TCP_KEEPCNT", count)):       # Python 3.6.5+ / Windows 10 1703 以降
        option = getattr(socket, name, None)
        if option is not None:                         # 非対応環境では定数自体が存在しない
            sock.setsockopt(socket.IPPROTO_TCP, option, value)

旧版の本記事は 3 つの定数をまとめて「Python 3.7 以降 / Windows 10 1709 以降」と書いていましたが、TCP_KEEPCNT だけ条件が違います（CPython の変更履歴では 3.6.5 で Windows 対応、Microsoft の対応表では Windows 10 バージョン 1703 から）。上のコードのコメントを修正済みです。

プローブ回数については、当サイトでは「設定できた」までしか確認できていません。Microsoft のドキュメントは 2 ページで食い違っています。SIO_KEEPALIVE_VALS のページは “On Windows Vista and later, the number of keep-alive probes … is set to 10 and cannot be changed”（10 回固定で変更不可）と書き、ソケットオプションの一覧ページは TCP_KEEPCNT が Windows 10 バージョン 1703 から設定可能と書いています。当サイトの実測では、TCP_KEEPCNT に 5 を設定して getsockopt で読み戻すと 5 が返りました（API はエラーなく受理）。ただし実際に送出されるプローブが 5 回になるかはパケットキャプチャが必要で、当サイトでは未検証です。回数に依存する設計は避けてください。

そして、より移植性が高いのはアプリ層の PINGです。30 秒おきに自前の PING 電文を送り、応答が来なければ切断とみなす。この方式なら OS もバージョンも問いませんし、「TCP はつながっているが機器のアプリが固まっている」状態も検知できます。TCP Keep-Alive は OS が返すため、相手のアプリが死んでいても応答してしまうことがあります。両方入れておくのが安全です。

切断に気づくまでの時間は、設定値から見積もれます。以下はいずれも設定値から計算した目安で、パケットの実送出やケーブル抜去からの実測は当サイトでは未検証です。

詰めた場合の値は、上の enable_tcp_keepalive の既定引数（idle=30／interval=5／count=4）から算出したものです。24 時間止められない設備で「何秒で切断に気づけるか」は、この計算で設計値として押さえられます（実ネットワークでの実測は今後の課題です）。

9. SO_REUSEADDR の罠 — Windows では意味が違う

ネット上のサーバーサンプルは、ほぼ例外なく setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) を入れています。本記事の旧版もそうでした。Windows のサーバーでは、この 1 行を消してください。

Linux でこれを付ける理由は「TIME_WAIT（接続を閉じた直後にしばらくポートが残る状態）の残っているポートに再 bind できるようにして、サーバーの再起動を速くする」ためです。Windows での意味は違います。Microsoft の公式ドキュメントは、こう書いています。

“The SO_REUSEADDR socket option allows a socket to forcibly bind to a port in use by another socket.”（他のソケットが使用中のポートへ強制的に bind できる）

“Once the second socket has successfully bound, the behavior for all sockets bound to that port is indeterminate.”（2 つ目が bind できてしまうと、そのポートに bind した全ソケットの挙動は不定になる）

“No special privileges are required to use this option.”（特別な権限は不要）／“All server applications must set SO_EXCLUSIVEADDRUSE for a strong level of socket security.”

9.1 実測 — 待ち受け中のポートは奪えるか（5 パターン）

当サイトの Windows 11 環境で、待ち受け中のポートへ別ソケットが bind できるかを 5 パターン試しました（すべて自 PC のループバック内・同一ユーザーで実施）。

読み方は 2 つです。①自分のサーバーに SO_REUSEADDR を付けた瞬間、同じ PC の別プロセスが同じポートを奪える状態になります。付けなければ、当環境では後続の bind は WinError 10013 で拒否されました。つまりこのオプションは、Windows では自分で自分の鍵を開けているのと同じです。

②bind(("", port))（全 NIC 待ち受け）は、オプションを何も付けなくても、より具体的なアドレスを指定した後発のソケットに接続を奪われました。当環境では、127.0.0.1 宛の 5 回の接続がすべて後から bind した側に渡っています。これを止められるのが SO_EXCLUSIVEADDRUSE でした。Windows でサーバーを常駐させるなら、次の 1 行を入れておくのが安全側です。

# ※断片です（HOST / PORT は定義済みの前提。動く形は 2.2 の server_simulator.py）
listener = socket.socket()
if hasattr(socket, "SO_EXCLUSIVEADDRUSE"):   # Windows でのみ定義される定数
    listener.setsockopt(socket.SOL_SOCKET, socket.SO_EXCLUSIVEADDRUSE, 1)
listener.bind((HOST, PORT))
listener.listen()

なお「乗っ取られたらどちらがデータを受け取るのか」は、Microsoft が “indeterminate”（不定）と明記しています。当環境で 2 つのソケットが同じ 127.0.0.1:port を SO_REUSEADDR 付きで待ち受けた場合、10 回の接続はすべて先に bind した側が受け取りました。ただしこれは当環境で観測された挙動であって、依存してよい仕様ではありません。

TIME_WAIT についても実測しました。サーバー側から接続を閉じて netstat に TIME_WAIT が残っている状態で、同じポートへオプション無しで bind と listen が成功しました（Windows 11・ループバック）。Linux では TIME_WAIT が残る間の bind が失敗するため SO_REUSEADDR を付ける習慣がありますが（当サイトでは Linux 側は未検証）、Windows でその目的のために付ける必要はなく、代わりに乗っ取りの穴だけが開きます。

9.2 Windows なら SO_EXCLUSIVEADDRUSE、Linux なら SO_REUSEADDR

Linux で同じコードを動かす場合は、逆に SO_REUSEADDR が必要になることがあります。hasattr で分岐して Windows なら SO_EXCLUSIVEADDRUSE、それ以外なら SO_REUSEADDR と書き分けるのが実務的です。

⚠️ この検証は自分の PC のループバック内でのみ行ってください。社内 LAN で他プロセスのポートを奪う実験をすると、業務システムを止めます。

10. Nagle と TCP_NODELAY

小さなコマンドを送って応答を待つ、リクエスト・レスポンス型の機器通信では Nagle アルゴリズムが引っかかることがあります。Nagle は「小さなデータをまとめてから送る」最適化で、送信の遅延と引き換えにパケット数を減らします。

Python の socket は既定で Nagle が有効です。当サイトの環境でも、新しいソケットの TCP_NODELAY は 0（＝Nagle 無効化していない）でした。Go や Node.js が既定で無効化しているのと違う点で、2026 年 1 月に CPython へ「既定で TCP_NODELAY を有効にする」提案（issue #144254）が出されましたが、not planned としてクローズされています。つまり現状、切りたければ自分で 1 行書くしかありません。

sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)   # 接続済みの sock に対して 1 行（断片。11 章の device_client.py には未組込＝必要な場合だけ足す）

効果については正直に書きます。当サイトのループバック実測では、差はほぼ出ませんでした。16 バイトのリクエストと 16 バイトの応答を 1000 往復させた実測は、TCP_NODELAY 無効で 0.028 秒・0.030 秒、有効で 0.026 秒・0.028 秒（1 往復あたり 0.026〜0.030 ミリ秒）。ループバックは実 NIC を通らず、Nagle と遅延 ACK の待ち合わせが起きにくいため、この結果から「効果が無い」とは言えません。実ネットワーク経由・実機器相手での効果は当サイトでは未検証です。「送信してから応答まで数十〜数百ミリ秒の説明できない遅延がある」ときの調査項目として覚えておく、という位置づけが妥当です。

11. 24 時間運用に耐える DeviceClient

ここまでの対策（フレーム境界・タイムアウト・切断検知・自動再接続・スレッド分離）を 1 つのクラスにまとめます。外から見える API は start / stop / send / recv_queue の 4 つだけです。

11.1 接続マネージャ

# device_client.py — 24 時間運用向けの接続マネージャ
import logging
import queue
import socket
import struct
import threading
from typing import Optional

logger = logging.getLogger(__name__)
HEADER = struct.Struct(">I")
MAX_FRAME = 1024 * 1024


def enable_tcp_keepalive(sock: socket.socket, idle: int = 30,
                         interval: int = 5, count: int = 4) -> None:
    """OS の TCP Keep-Alive を有効にし、間隔を詰める（8 章）。"""
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
    for name, value in (("TCP_KEEPIDLE", idle),
                        ("TCP_KEEPINTVL", interval),
                        ("TCP_KEEPCNT", count)):
        option = getattr(socket, name, None)
        if option is not None:
            sock.setsockopt(socket.IPPROTO_TCP, option, value)


class DeviceClient:
    """産業機器との TCP 接続を管理するクライアント。

    設計方針:
        - 切断は例外ではなく状態として扱う（自動再接続）
        - 受信は専用スレッド、データは queue 経由で外に渡す
        - 外部からは start / stop / send だけ呼べばよい
    """

    def __init__(self, host: str, port: int,
                 connect_timeout: float = 5.0,
                 recv_timeout: float = 5.0,
                 backoff_start: float = 1.0,
                 backoff_max: float = 30.0):
        self.host = host
        self.port = port
        self.connect_timeout = connect_timeout
        self.recv_timeout = recv_timeout
        self.backoff_start = backoff_start
        self.backoff_max = backoff_max
        self.recv_queue: "queue.Queue[bytes]" = queue.Queue(maxsize=1000)
        self._sock: Optional[socket.socket] = None
        self._stop_event = threading.Event()
        self._thread: Optional[threading.Thread] = None
        self._send_lock = threading.Lock()

    def start(self) -> None:
        self._stop_event.clear()
        self._thread = threading.Thread(target=self._run, daemon=True)
        self._thread.start()

    def stop(self) -> None:
        self._stop_event.set()
        self._close_socket()
        if self._thread:
            self._thread.join(timeout=5)

    def send(self, payload: bytes) -> None:
        with self._send_lock:
            sock = self._sock       # 再接続スレッドとの競合に備えてローカル変数に取る
            if sock is None:
                raise ConnectionError("not connected")
            sock.sendall(HEADER.pack(len(payload)) + payload)

    def _run(self) -> None:
        backoff = self.backoff_start
        while not self._stop_event.is_set():
            try:
                self._connect()
                backoff = self.backoff_start   # つながったら待ち時間を戻す
                self._recv_loop()
            except (ConnectionError, OSError) as e:   # TimeoutError も OSError の一種
                logger.warning("connection error: %s", e)
            finally:
                self._close_socket()
            if self._stop_event.is_set():
                break
            logger.info("reconnect in %.1fs", backoff)
            if self._stop_event.wait(backoff):        # stop() されたら即座に抜ける
                break
            backoff = min(backoff * 2, self.backoff_max)

    def _connect(self) -> None:
        sock = socket.create_connection((self.host, self.port),
                                        timeout=self.connect_timeout)
        sock.settimeout(self.recv_timeout)   # 接続後は受信のタイムアウトに切り替える
        enable_tcp_keepalive(sock)
        self._sock = sock
        logger.info("connected to %s:%s", self.host, self.port)

    def _recv_loop(self) -> None:
        while not self._stop_event.is_set():
            frame = self._recv_frame()
            try:
                self.recv_queue.put_nowait(frame)
            except queue.Full:
                logger.warning("recv_queue is full, dropping frame")

    def _recv_frame(self) -> bytes:
        length, = HEADER.unpack(self._recv_exact(HEADER.size))
        if length > MAX_FRAME:
            raise ConnectionError(f"frame too large: {length}")
        return self._recv_exact(length)

    def _recv_exact(self, n: int) -> bytes:
        sock = self._sock           # stop() との競合に備えてローカル変数に取る
        if sock is None:
            raise ConnectionError("not connected")
        buf = bytearray()
        while len(buf) < n:
            chunk = sock.recv(n - len(buf))
            if not chunk:
                raise ConnectionError("peer closed")
            buf.extend(chunk)
        return bytes(buf)

    def _close_socket(self) -> None:
        if self._sock:
            try:
                self._sock.close()
            except OSError:
                pass
            self._sock = None

旧版から変えた点は 3 つです。①再接続の待ち時間を指数バックオフにした（機器が長時間停止しているとき、1 秒ごとに接続を試み続けると機器側の接続数制限に当たります）。②time.sleep ではなく _stop_event.wait(backoff) で待つ（停止指示が来たら 30 秒待ちの途中でも即座に抜けられます）。③接続直後に keep-alive を有効化（8 章）。

send() は切断中に ConnectionError / OSError を投げるため、呼び出し側で捕捉してください。3 章で触れたとおり sendall は「エラー時に何バイト送れたか分からない」ので、再送する場合は同じコマンドを 2 回受けても害の無い設計が前提になります。

1 つ重要な設計前提があります。この実装は recv_timeout（既定 5 秒）の間に何も受信しないと切断とみなして再接続します。つまり定周期でデータを送ってくる機器向けの設計です。「異常時だけ通知してくる」ようなイベント駆動型の機器に対しては、健全な接続を 5 秒ごとに切断・再接続し続けてしまいます（同時接続数に制限のある機器では、この再接続連打が原因で接続拒否に陥る機種もあります）。イベント駆動型の機器では、recv_timeout を十分長くするか、8 章のアプリ層 PING を併用して「無通信＝異常」と判定できるようにしてください。

11.2 使う側のコード

# main.py — DeviceClient を使う側
import logging
import queue

from device_client import DeviceClient

logging.basicConfig(level=logging.INFO,
                    format="%(asctime)s %(levelname)s %(message)s")

client = DeviceClient("127.0.0.1", 9000)
client.start()

try:
    while True:
        try:
            frame = client.recv_queue.get(timeout=1.0)
        except queue.Empty:
            continue          # 受信が無い間も UI やタイマ処理を回せる
        print(f"frame: {frame!r}")
except KeyboardInterrupt:
    pass
finally:
    client.stop()

UI（tkinter / PyQt / Streamlit）から使う場合は、UI 側の定期コールバックで recv_queue.get_nowait() を呼ぶ設計にすると、画面はノンブロッキングで動き続けます。組み合わせた完成形は tkinter 監視アプリの記事にあります。socket に触るスレッドは 1 本だけ——この原則を崩すと、送受信が混ざって再現しない不具合になります。

11.3 定周期送信するシミュレータと、再接続の実測

# server_with_length_prefix.py — 長さプレフィックス方式で定周期送信する機器シミュレータ
import socket
import struct
import time

HOST, PORT = "127.0.0.1", 9000
HEADER = struct.Struct(">I")

listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
if hasattr(socket, "SO_EXCLUSIVEADDRUSE"):
    listener.setsockopt(socket.SOL_SOCKET, socket.SO_EXCLUSIVEADDRUSE, 1)

with listener:
    listener.bind((HOST, PORT))
    listener.listen()
    print(f"listening on {HOST}:{PORT}")
    while True:
        conn, addr = listener.accept()
        with conn:
            print(f"connected: {addr}")
            counter = 0
            try:
                while True:
                    payload = f"DATA-{counter}".encode("utf-8")
                    conn.sendall(HEADER.pack(len(payload)) + payload)
                    counter += 1
                    time.sleep(0.5)
            except OSError as e:
                # ConnectionReset / ConnectionAborted / BrokenPipe はすべて OSError の一種
                print(f"disconnected: {e}")

シミュレータを起動して main.py を動かすと、0.5 秒ごとに DATA-0, DATA-1, ... が流れます。シミュレータを Ctrl+C で落として数秒後に再起動すると、当サイトの実測ではこうなりました。

2026-08-31 19:22:19,300 INFO connected to 127.0.0.1:9000
2026-08-31 19:22:22,225 WARNING connection error: [WinError 10054] 既存の接続はリモート ホストに強制的に切断されました。
2026-08-31 19:22:22,225 INFO reconnect in 1.0s
2026-08-31 19:22:25,273 WARNING connection error: [WinError 10061] 対象のコンピューターによって拒否されたため、接続できませんでした。
2026-08-31 19:22:25,273 INFO reconnect in 2.0s
2026-08-31 19:22:27,302 INFO connected to 127.0.0.1:9000

切断を検知してから復帰まで 5.1 秒。内訳は「1 秒待って 1 回目の再試行 → 拒否が返るまで約 2 秒（7 章）→ 2 秒待って 2 回目で成功」です。バックオフの待ち時間だけでなく、接続失敗が返るまでの時間も復帰時間に乗ることが、この実測から読み取れます。

なお、複数の機器を 1 つの Python プロセスで並列に監視する構成は、selectors と asyncio のどちらに寄せるかという設計判断が絡む別トピックのため、本記事では扱いません。1 機器につき 1 本の接続で足りるなら、ここまでの DeviceClient で必要なものは揃っています。

12. デバッグの定石

「つながらない」ときに最初にやることは、Python を疑う前にポートの状態を見ることです。当サイトの環境で、ポート 9000 を待ち受けている状態の実出力です。

PS> netstat -ano | findstr :9000
  TCP         127.0.0.1:9000         0.0.0.0:0              LISTENING       21564

PS> Get-NetTCPConnection -LocalPort 9000 | Select-Object LocalAddress,LocalPort,State,OwningProcess

LocalAddress LocalPort  State OwningProcess
------------ ---------  ----- -------------
127.0.0.1         9000 Listen         21564

最後の列がプロセス ID です。Get-Process -Id 21564 で犯人が分かります。WinError 10048（アドレス使用中）が出たときは、まずこれで「前回の自分」が残っていないかを確認してください。デバッグ実行を止め損ねたプロセスが居座っているケースが大半です。

切り分けは次の順で進めます。①ping で機器まで届くかを確認します（届かなければネットワーク側の問題で、Python の出番ではありません）。②ポート番号を仕様書で再確認します（ConnectionRefusedError の多くはこれです）。③自分のダミーサーバーに同じコードで接続できるか試します（できるならクライアント側のコードは無罪で、機器側の設定や接続元 IP 制限を疑います）。④受信した生バイト列を repr() でそのままログに残します（仕様書と 1 バイトずつ突き合わせます。5 章のバイト数ずれはここで見つかります）。

パケットキャプチャ（Wireshark 等）はさらに強力ですが、当サイトでは本記事の検証に使用していません。また、業務ネットワークでのキャプチャは会社の情報セキュリティ規程に触れることがあるため、実施前に必ず確認してください。

13. 性能はどこで頭打ちになるか（実測）

「Python の socket は遅いのでは」という質問をよく受けます。当サイトのループバック環境で測った限り、ボトルネックは socket ではありませんでした。

すべて同一 PC 内のループバックでの値です。実 NIC・スイッチを経由した通信では、伝送時間と OS の割り込み処理が加わるため、この数値は出ません。機器からの送信が毎秒数件〜数百フレームに収まる用途であれば、標準 socket そのものが先に音を上げる場面は考えにくい——という参考値として読んでください。

実測から読み取れる設計上の示唆が 3 つあります。①queue の上限は必ず設ける——消費側が止まると 0.014 秒で 1000 件が埋まりました。上限なしにしていたら、そのままメモリを食い続けます。②遅くなる原因は socket の外側にある——1 フレーム 5.6 マイクロ秒に対し、UI 描画・ファイル IO・pandas 集計は桁が 3 つ以上違います。受信スレッドから直接それらを呼ばず、queue 越しに渡してください。③高頻度のコンソール出力は疑う価値がある——パイプへの print は 1 回 1.4 マイクロ秒でしたが、実際のコンソールへの描画はこれより重くなります（当サイト未計測）。常用ログは logging のファイル出力にして、デバッグ用の print は残さないのが無難です。

それでも足りない場合（毎秒 1 万フレーム超が続く、接続数が数百に増える）は、asyncio への移行や、フレームをまとめてバッチ処理する設計の検討に入ります。

14. 実機につなぐときの安全ルール

シミュレータで動きを確認できたら、実機への切り替えは HOST を機器の IP アドレスに変えるだけです（IP とポートは機器の仕様書または設定画面で確認し、まず ping で疎通を確かめてから接続します）。ただしその前に、必ず守ってほしいことが 3 つあります。

• 制御ネットワークに PC を勝手につながない: IP アドレスの重複は設備停止に直結します。IP の払い出しと接続の可否は、設備管理責任者・情報システム部門に必ず確認を
• 最初の接続テストは計画停止・保全日に: 稼働中のラインで初接続を試さない。読み出しだけのつもりでも、機器側の同時接続数を超えれば既存の接続が切られることがあります
• 閉域ネットワーク内でのみ使う: 本記事の通信は平文・認証なしの前提です。事務 LAN やインターネットに露出する経路では使わないでください

9 章の SO_REUSEADDR の検証も同じです。ポートの奪い合いを試すのは自分の PC のループバック内だけにしてください。

15. 24 時間運用チェックリスト

16. おわりに

TCP/IP 通信は「動かす」のと「24 時間止まらず動かし続ける」の間に、見えにくいギャップがあります。本記事で埋めたのは、境界（4・5 章）・時間（6 章）・失敗の識別（7 章）・切断（8 章）・ポートの安全（9 章）・復帰（11 章）の 6 点です。DeviceClient は現場で使えるサイズの最小骨格で、実プロジェクトではここに「フレーム種別ごとのパーサ」「メトリクス収集」「監視アラート連携」を足していくことになります。

本記事の実測はすべてループバックでの値であり、実機器・実ネットワークでの検証は行っていません。それでも、「1 回の send が 1 回の recv で届くとは限らない」「@ を書くとバイト数がずれる」「Windows の SO_REUSEADDR は乗っ取りを許す」——この 3 つは環境を問わず効く知識です。手元のダミーサーバーで再現できるので、実機が届く前に一度試してみてください。