Python のソケットは、ネットワーク通信のための強力なツールであり、異なるシステム間でメッセージを送受信することを可能にします。このチュートリアルでは、Python のソケットを使用して接続を確立し、データを送信し、応答を受信するプロセスを説明します。この実験（Lab）の終わりには、ネットワークを介して互いに通信できるクライアントアプリケーションとサーバーアプリケーションの両方を構築できるようになります。
この実践的な知識は、チャットプログラムから分散システムまで、より複雑なネットワークアプリケーションを開発するための基盤となります。
まず、ソケットとは何か、そして Python でどのように機能するのかを理解することから始めましょう。次に、最初のソケットを作成し、それがどのように初期化されるかを確認します。
ソケットは、ネットワークを介してデータを送受信するためのエンドポイントです。これらは、ネットワーク通信へのプログラミングインターフェースを提供し、アプリケーションが場所に関係なく情報を交換できるようにします。
ネットワークプログラミングでは、通常、クライアントサーバーモデルを使用します。
Python の組み込み socket モジュールを使用すると、ネットワークプロトコルの複雑な詳細をすべて理解しなくても、ソケットを簡単に操作できます。
最も一般的な 2 つのソケットの種類は次のとおりです。
この実験（Lab）では、信頼性の高い通信を必要とするアプリケーションで最も一般的に使用される TCP ソケットに焦点を当てます。
ソケットを初期化する簡単な Python スクリプトを作成しましょう。WebIDE を開き、次の手順に従います。
mkdir -p ~/project/socket_lab
cd ~/project/socket_labsocket_basics.py という新しい Python ファイルを作成します。WebIDE で、「New File」ボタンをクリックするか、「File」メニューを使用して「New File」を選択し、socket_lab ディレクトリ内に socket_basics.py という名前を付けます。
import socket
## Creating a socket
print("Creating a new socket...")
my_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
print(f"Socket created: {my_socket}")
## Explaining the parameters:
print("\nSocket parameters explained:")
print("AF_INET: Using IPv4 addressing")
print("SOCK_STREAM: Using TCP protocol for reliable data transmission")
## Getting available socket methods
print("\nSome methods available on socket objects:")
methods = [method for method in dir(my_socket) if not method.startswith('_')]
print(', '.join(methods[:10]) + '...') ## Showing first 10 methods
## Closing the socket
my_socket.close()
print("\nSocket closed")ファイルを保存します。
スクリプトを実行して出力を確認します。
python3 ~/project/socket_lab/socket_basics.py次のような出力が表示されるはずです。
Creating a new socket...
Socket created: <socket.socket fd=3, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=0, laddr=('0.0.0.0', 0)>
Socket parameters explained:
AF_INET: Using IPv4 addressing
SOCK_STREAM: Using TCP protocol for reliable data transmission
Some methods available on socket objects:
accept, bind, close, connect, connect_ex, detach, fileno, getpeername, getsockname, getsockopt...
Socket closedこの出力は、ソケットオブジェクトを正常に作成し、そのプロパティの一部を調べたことを示しています。次のセクションでは、ソケットを使用して、互いに通信できるサーバーとクライアントアプリケーションを構築します。
次に進む前に、使用する主要なソケット関数を理解しましょう。
socket() - 新しいソケットオブジェクトを作成します。bind() - ソケットを特定のネットワークインターフェースとポートに関連付けます。listen() - サーバーが接続を受け入れるようにします。accept() - クライアントからの接続を受け入れます。connect() - リモートアドレスに接続します。send() - 接続されたソケットにデータを送信します。recv() - 接続されたソケットからデータを受信します。close() - ソケットを閉じます。次のステップでは、これらの関数を使用してサーバーを作成します。
ソケットの基本を理解したので、接続をリッスンし、クライアントからメッセージを受信するシンプルなサーバーを作成しましょう。
ソケットサーバーは、次の一般的な手順に従います。
このパターンを Python で実装しましょう。
socket_lab ディレクトリに、server.py という新しい Python ファイルを作成します。WebIDE で、「New File」ボタンをクリックするか、「File」メニューを使用して「New File」を選択し、socket_lab ディレクトリ内に server.py という名前を付けます。
import socket
def start_server():
## Server configuration
host = '127.0.0.1' ## localhost
port = 12345 ## arbitrary non-privileged port
## Create socket
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
## Set socket option to reuse address (helps avoid "Address already in use" errors)
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
## Bind socket to address and port
server_socket.bind((host, port))
## Listen for connections (queue up to 5 connection requests)
server_socket.listen(5)
print(f"Server started on {host}:{port}")
print("Waiting for client connection...")
try:
## Accept connection
client_socket, client_address = server_socket.accept()
print(f"Connection established with {client_address}")
## Receive data from client
data = client_socket.recv(1024) ## receive up to 1024 bytes
print(f"Message received: {data.decode()}")
## Send response to client
response = "Message received by server"
client_socket.send(response.encode())
## Close client connection
client_socket.close()
except KeyboardInterrupt:
print("\nServer shutting down...")
finally:
## Close server socket
server_socket.close()
print("Server socket closed")
if __name__ == "__main__":
start_server()サーバーの主要なコンポーネントを分解してみましょう。
ソケットの作成: socket.socket() を使用して、AF_INET (IPv4) と SOCK_STREAM (TCP プロトコル) で TCP ソケットを作成します。
ソケットオプション: アドレスを再利用するオプションを設定します。これは、シャットダウン後にサーバーをすばやく再起動する際に「Address already in use」エラーを回避するのに役立ちます。
バインディング: ソケットをアドレス 127.0.0.1 (localhost) とポート 12345 にバインドします。これにより、特定のネットワークロケーションで接続を受信することをオペレーティングシステムに指示します。
リスニング: listen(5) の呼び出しは、新しい接続を拒否する前に、ソケットに最大 5 つの接続要求をキューに入れるように指示します。
接続の受け入れ: accept() メソッドは、クライアントが接続するまでブロック (待機) し、そのクライアントと通信するための新しいソケットオブジェクトとクライアントのアドレスを返します。
データの受信: recv(1024) を使用して、クライアントから最大 1024 バイトのデータを受信します。データはバイトとして送信されるため、decode() を使用して文字列に変換します。
データの送信: send() メソッドを使用して、クライアントに応答を返します。送信する前に、文字列を encode() してバイトに変換します。
クローズ: 最後に、クライアントソケットとサーバーソケットの両方を閉じて、リソースを解放します。
サーバーを実行して、実際に動作していることを確認しましょう。まだクライアントを作成していないため、あまり多くのことは行われませんが、正しく起動することを確認できます。
python3 ~/project/socket_lab/server.py次のような出力が表示されるはずです。
Server started on 127.0.0.1:12345
Waiting for client connection...サーバーは、クライアントが接続するのを待機しています。まだクライアントがないため、Ctrl+C を押してサーバーを停止します。
^C
Server shutting down...
Server socket closed次のステップでは、サーバーに接続するクライアントを作成します。
これでサーバーができたので、それに接続してメッセージを送信できるクライアントを作成する必要があります。シンプルなクライアントアプリケーションを構築しましょう。
ソケットクライアントは、次の一般的な手順に従います。
socket_lab ディレクトリに、client.py という新しい Python ファイルを作成します。WebIDE で、「New File」ボタンをクリックするか、「File」メニューを使用して「New File」を選択し、socket_lab ディレクトリ内に client.py という名前を付けます。
import socket
def start_client():
## Server information to connect to
host = '127.0.0.1' ## localhost - same as server
port = 12345 ## same port as server
try:
## Create socket
client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
## Connect to server
print(f"Connecting to server at {host}:{port}...")
client_socket.connect((host, port))
print("Connected to server")
## Send a message
message = "Hello from the client!"
print(f"Sending message: {message}")
client_socket.send(message.encode())
## Receive response
response = client_socket.recv(1024)
print(f"Response from server: {response.decode()}")
except ConnectionRefusedError:
print("Connection failed. Make sure the server is running.")
except Exception as e:
print(f"An error occurred: {e}")
finally:
## Close socket
client_socket.close()
print("Connection closed")
if __name__ == "__main__":
start_client()クライアントコードの主要な部分を調べてみましょう。
ソケットの作成: サーバーと同様に、socket.socket() を使用して TCP ソケットを作成します。
接続: バインドとリスニングの代わりに、クライアントは connect() を使用して、指定されたホストとポートでサーバーへの接続を確立します。
データの送信: send() メソッドを使用してメッセージを送信し、文字列をバイトにエンコードしていることを確認します。
データの受信: recv(1024) を使用して、サーバーの応答を受信し、文字列にデコードし直します。
エラー処理: サーバーが利用できない場合 (ConnectionRefusedError) など、一般的な問題をキャッチするためのエラー処理を含めます。
クローズ: リソースを解放するために、完了したらソケットを閉じます。
次に、クライアントとサーバーを一緒にテストしましょう。これらは、別々のターミナルウィンドウで実行する必要があります。
python3 ~/project/socket_lab/server.py次のように表示されるはずです。
Server started on 127.0.0.1:12345
Waiting for client connection...python3 ~/project/socket_lab/client.pyクライアントターミナルには、次のような出力が表示されるはずです。
Connecting to server at 127.0.0.1:12345...
Connected to server
Sending message: Hello from the client!
Response from server: Message received by server
Connection closedそして、サーバーターミナルには、次のように表示されるはずです。
Connection established with ('127.0.0.1', 55234) ## The port number may differ
Message received: Hello from the client!クライアントが切断した後、現在の実装では単一の接続しか処理しないため、サーバーは停止します。まだ実行中の場合は、サーバーターミナルで Ctrl+C を押してシャットダウンします。
これは、Python ソケットを使用したクライアントとサーバー間の通信の成功を示しています。クライアントはサーバーにメッセージを送信でき、サーバーはそれを受信して応答を返すことができます。
現在のサーバーとクライアントの実装では、閉じる前に単一のメッセージ交換しか処理できません。ほとんどの現実世界のアプリケーションでは、接続を維持し、複数のメッセージを処理する必要があります。よりインタラクティブなエクスペリエンスを作成するために、コードを強化しましょう。
まず、サーバーを変更して、接続を継続的に受け入れ、各クライアントからの複数のメッセージを処理するようにしましょう。
server.py ファイルを開き、コードを次のように置き換えます。import socket
def start_server():
## Server configuration
host = '127.0.0.1'
port = 12345
## Create socket
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
try:
## Bind and listen
server_socket.bind((host, port))
server_socket.listen(5)
print(f"Server running on {host}:{port}")
print("Press Ctrl+C to stop the server")
while True: ## Continuous server loop
print("\nWaiting for a connection...")
client_socket, client_address = server_socket.accept()
print(f"Connected to client: {client_address}")
## Handle client communication
handle_client(client_socket)
except KeyboardInterrupt:
print("\nServer is shutting down...")
finally:
server_socket.close()
print("Server closed")
def handle_client(client_socket):
try:
while True: ## Keep receiving messages until client disconnects
## Receive data
data = client_socket.recv(1024)
## If no data, client has disconnected
if not data:
break
received_message = data.decode()
print(f"Received: {received_message}")
## Process the message (in this case, just echo it back with a prefix)
response = f"Server received: {received_message}"
client_socket.send(response.encode())
except Exception as e:
print(f"Error handling client: {e}")
finally:
## Close client socket
client_socket.close()
print("Client connection closed")
if __name__ == "__main__":
start_server()次に、ユーザーが複数のメッセージを送信できるインタラクティブなクライアントを作成しましょう。
client.py ファイルを開き、コードを次のように置き換えます。import socket
def start_client():
## Server information
host = '127.0.0.1'
port = 12345
try:
## Create socket and connect
client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
print(f"Connecting to server at {host}:{port}...")
client_socket.connect((host, port))
print("Connected to server!")
## Interactive message sending
while True:
## Get message from user
message = input("\nEnter message to send (or 'quit' to exit): ")
## Check if user wants to quit
if message.lower() == 'quit':
print("Closing connection...")
break
## Send message
client_socket.send(message.encode())
## Receive response
response = client_socket.recv(1024)
print(f"Response from server: {response.decode()}")
except ConnectionRefusedError:
print("Connection failed. Make sure the server is running.")
except Exception as e:
print(f"An error occurred: {e}")
finally:
## Close connection
try:
client_socket.close()
except:
pass
print("Disconnected from server")
if __name__ == "__main__":
start_client()while True ループを追加しましたhandle_client 関数を作成しましたif not data:) をチェックします。これは、クライアントが切断されたことを示しますwhile True ループを追加しました強化されたサーバーとクライアントをテストしましょう。
python3 ~/project/socket_lab/server.py次のように表示されるはずです。
Server running on 127.0.0.1:12345
Press Ctrl+C to stop the server
Waiting for a connection...python3 ~/project/socket_lab/client.py次のように表示されるはずです。
Connecting to server at 127.0.0.1:12345...
Connected to server!
Enter message to send (or 'quit' to exit):Enter message to send (or 'quit' to exit): Hello, server!
Response from server: Server received: Hello, server!
Enter message to send (or 'quit' to exit):Connected to client: ('127.0.0.1', 59042)
Received: Hello, server!
Received: This is another messageEnter message to send (or 'quit' to exit): quit
Closing connection...
Disconnected from serverClient connection closed
Waiting for a connection...これらの強化により、Python ソケットを使用して、より現実的でインタラクティブなクライアントサーバー通信システムを作成しました。
現実世界のアプリケーションでは、サーバーは通常、複数のクライアントを同時に処理し、さまざまなエラー状態を適切に管理する必要があります。これらの考慮事項を念頭に置いて、実装を改善しましょう。
複数のクライアントを同時に処理するには、いくつかの方法があります。
この実験では、理解と実装が比較的簡単なスレッディングベースのアプローチを実装します。
スレッドを使用して複数のクライアントを処理するように、サーバーを変更しましょう。
server.py ファイルを開き、コードを次のように置き換えます。import socket
import threading
def handle_client(client_socket, client_address):
"""Handle communication with a single client"""
try:
print(f"[NEW CONNECTION] {client_address} connected.")
while True:
## Receive client data
try:
data = client_socket.recv(1024)
if not data:
break ## Client disconnected
message = data.decode()
print(f"[{client_address}] {message}")
## Send response
response = f"Message '{message}' received successfully"
client_socket.send(response.encode())
except ConnectionResetError:
print(f"[{client_address}] Connection reset by client")
break
except Exception as e:
print(f"[ERROR] {e}")
finally:
## Clean up when client disconnects
client_socket.close()
print(f"[DISCONNECTED] {client_address} disconnected")
def start_server():
"""Start the server and listen for connections"""
## Server configuration
host = '127.0.0.1'
port = 12345
## Create socket
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
## Set socket option to reuse address
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
try:
## Bind to host and port
server_socket.bind((host, port))
## Listen for connections
server_socket.listen(5)
print(f"[STARTING] Server is listening on {host}:{port}")
while True:
## Accept client connection
client_socket, client_address = server_socket.accept()
## Create a new thread to handle the client
client_thread = threading.Thread(
target=handle_client,
args=(client_socket, client_address)
)
client_thread.daemon = True ## Thread will close when main program exits
client_thread.start()
## Display active connections
print(f"[ACTIVE CONNECTIONS] {threading.active_count() - 1}")
except KeyboardInterrupt:
print("\n[SHUTTING DOWN] Server is shutting down...")
except Exception as e:
print(f"[ERROR] {e}")
finally:
server_socket.close()
print("[CLOSED] Server socket closed")
if __name__ == "__main__":
start_server()また、より良いエラー処理でクライアントを強化しましょう。
client.py ファイルを開き、コードを次のように置き換えます。import socket
import sys
import time
def start_client():
"""Start a client that connects to the server"""
## Server information
host = '127.0.0.1'
port = 12345
## Create socket
client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
## Set a timeout for connection attempts (5 seconds)
client_socket.settimeout(5)
try:
## Connect to server
print(f"[CONNECTING] Connecting to server at {host}:{port}...")
client_socket.connect((host, port))
## Reset timeout to none for regular communication
client_socket.settimeout(None)
print("[CONNECTED] Connected to server")
## Communication loop
while True:
## Get user input
message = input("\nEnter message (or 'quit' to exit): ")
if message.lower() == 'quit':
print("[CLOSING] Closing connection by request...")
break
try:
## Send message
client_socket.send(message.encode())
## Wait for response
response = client_socket.recv(1024)
print(f"[RESPONSE] {response.decode()}")
except ConnectionResetError:
print("[ERROR] Connection was reset by the server")
break
except ConnectionAbortedError:
print("[ERROR] Connection was aborted")
break
except Exception as e:
print(f"[ERROR] {e}")
break
except socket.timeout:
print("[TIMEOUT] Connection attempt timed out. Is the server running?")
except ConnectionRefusedError:
print("[REFUSED] Connection refused. Make sure the server is running.")
except KeyboardInterrupt:
print("\n[INTERRUPT] Client shutting down...")
except Exception as e:
print(f"[ERROR] {e}")
finally:
## Close socket
try:
client_socket.close()
print("[DISCONNECTED] Disconnected from server")
except:
pass
if __name__ == "__main__":
start_client()threading モジュールを追加しました改善されたアプリケーションをテストしましょう。
python3 ~/project/socket_lab/server.py次のように表示されるはずです。
[STARTING] Server is listening on 127.0.0.1:12345python3 ~/project/socket_lab/client.pypython3 ~/project/socket_lab/client.py[NEW CONNECTION] ('127.0.0.1', 59124) connected.
[ACTIVE CONNECTIONS] 1
[NEW CONNECTION] ('127.0.0.1', 59126) connected.
[ACTIVE CONNECTIONS] 2[('127.0.0.1', 59124)] Hello from client 1
[('127.0.0.1', 59126)] Hello from client 2また、サーバーが実行されていない場合に何が起こるかをテストしましょう。
サーバーが停止していることを確認します (実行中の場合は Ctrl+C を押します)
クライアントを実行してみます。
python3 ~/project/socket_lab/client.py次のように表示されるはずです。
[CONNECTING] Connecting to server at 127.0.0.1:12345...
[REFUSED] Connection refused. Make sure the server is running.
[DISCONNECTED] Disconnected from serverクライアントは、エラーを適切に処理し、サーバーが実行されていない可能性があることをユーザーに通知するようになりました。
これらの強化により、複数のクライアントとさまざまなエラー状態を処理できる、堅牢なクライアントサーバーシステムを作成しました。これは、より複雑なネットワークアプリケーションを開発するための確固たる基盤です。
Python ソケットプログラミングに関するこの実験を完了したことをおめでとうございます。次の方法を正常に学習しました。
これらのスキルは、ネットワークプログラミングの基礎を形成し、シンプルなチャットプログラムから複雑な分散システムまで、幅広いネットワークアプリケーションの構築に適用できます。
学習を続けるには、以下を検討してください。
Python のソケットプログラミング機能は、ネットワークアプリケーション開発に最適であり、他の言語ではほとんど実現できない、シンプルさとパワーのバランスを提供します。
