자바 클라이언트에서 파이썬 서버로 멀티쓰레드 소켓 프로그래밍

안녕하세요.

 

오늘은 자바클라이언트에서 파이썬 소켓서버를 로컬에서 실서버까지 구현한 내용을 공유하겠습니다.

 

동시에 여러접속자의 처리를 구현하기 위해 멀티쓰레드 환경에서 처리하도록 구현하였습니다.

 

 

프로그래밍 기능

- 자바 클라이언트에서 String을 파이썬 소켓 서버로 전달하면 파이썬 소켓 서버는 클라이언트가 보낸 String을

  토대로 문서파일을 생성하여 자바로 파일 전달

 

운영환경

- 자바 클라인트(ec2 amzn_linux 운영체제)

- 파이썬 소켓 서버(ec2 windows 운영체제 )

 

 

 

소켓 프로그래밍 흐름

 

 

위 그림은 소켓 프로그래밍의 실행 흐름입니다.

 

소켓 서버에서는 소켓을 생성하고 자신의 서버 IP와 소켓 프로그램을 실행시킬 port번호를 결합(bind) 합니다.

 

그리고 listen() 메서드를 통해 클라이언트의 요청이 수신되는지 확인합니다. 

 

accept()를 통해 통신을 연결하고 send() / recv()로 data를 주고 받습니다. 

 

모든 데이터를 주고 받으면 클라이언트와 통신을 닫습니다. 

 

위 그림에서처럼 server socket은 listen()을 통해 추가적으로 들어오는 클라이언트의 요청에 계속 수신할 수 있습니다.

 

클리이언트는 더욱 간단합니다. 클라이언트는 연결하려는 소켓 서버의 ip와 port번호를 적은 소켓을 생성하고 연결하여 

 

send() / recv()로 소켓 서버와 데이터를 주고 받습니다.

 

소켓 통신은 통신하려는 두 서버를 연결하고 데이터를 주고 받는다고 생각하면 될 것 같습니다.

 

소켓 통신은 주로 채팅이나 실행환경이 다른 두 서버간에 통신을 위해 사용되는데 

 

http통신과 다른 점은 양방향 통신이 가능하다는 것입니다. 

 

http통신의 경우 클라이언트의 요청이 들어온 경우에만 서버가 응답을 줄 수 있는 반면

 

소켓 통신은 클라이언트의 요청 없이도 서버가 데이터를 전달해 줄 수 있습니다.

 

채팅과 같이 내가 채팅방에 입장하면 새로고침이나 어떤 액션을 취하지 않아도

 

상대방의 메세지가 내 화면에 표시되는 것이 양방향 통신의 예입니다.

 

제가 예시를 들 소스는 채팅도 아니고 양방향 통신을 위한 대표적인 예도 아니기 때문에

 

양방향 통신에 대한 설명은 여기까지 하고 소켓 통신의 연결과 데이터 송수신 과정을 소스를 통해 설명을 드리겠습니다.

 

 

 

- 자바 클라이언트 소스

import java.io.BufferedInputStream;
import java.io.DataInputStream;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.PrintStream;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.util.Random;
import java.util.Scanner;

public class ClientConnect {
	
    private Socket socket = null;
    private FileOutputStream fos = null;
    private DataInputStream din = null;
    private PrintStream pout = null;
    private Scanner scan = null;

    public ClientConnect(String customSocketIp, int portNumber) throws IOException {
        socket=new Socket(customSocketIp, portNumber);  
        scan = new Scanner(System.in);
        din = new DataInputStream(socket.getInputStream());
        pout = new PrintStream(socket.getOutputStream());
    }

    public void send(String msg) throws IOException {
    	byte[] data = msg.getBytes();
    	ByteBuffer b = ByteBuffer.allocate(4);
        //최초의 4바이트는 데이터 크기
        b.order(ByteOrder.LITTLE_ENDIAN);
        b.putInt(data.length);
        pout.write(b.array(), 0, 4);
        pout.print(msg);
        pout.flush();
    }

    public String recv() throws IOException {
        byte[] bytes = new byte[1024];
        din.read(bytes);
        String reply = new String(bytes, "UTF-8");
        return reply;
    }

    public void closeConnections() throws IOException {
        // Clean up when a connection is ended
        socket.close();
        din.close();
        pout.close();
        scan.close();
    }

    // Request a specific file from the server
    public String getFile(String path, String jsonStr) {
    	Random random1 = new Random();
    	long currentTime1 = System.currentTimeMillis();
		int randomValue1 = random1.nextInt(100);

		String newFileName = Long.toString(currentTime1) + "_"+randomValue1+"_mycustom.hwp";
        try {
            File file = new File(path, newFileName);
            // Create new file if it does not exist
            // Then request the file from server
            if(!file.exists()){
                file.createNewFile();
            }
            fos = new FileOutputStream(file);
            //보낼 데이터의 크기를 먼저 보낸다.
	       //데이터를 보낸다.
            send(jsonStr);

            // Get content in bytes and write to a file
            byte[] buffer = new byte[8192];
            for(int counter=0; (counter = din.read(buffer, 0, buffer.length)) >= 0;) {
                    fos.write(buffer, 0, counter);
            }
            fos.flush();
            fos.close();

        } catch (IOException e) {
            e.printStackTrace();
        }
        return newFileName;
    }
    
}

 

 

실행 소스

ClientConnect cc = new ClientConnect("소켓 서버 ip 번호", 5555);	
String newFileName= cc.getFile(path, myCustonJsonDto.getJsonString());
cc.closeConnections();

 

 

클라이언트 소켓을 생성하는 생성자에 서버 IP, 소켓 서버 실행 프로그램의 포트번호를 적어 소켓을 생성하고

 

예제 소스는 json String을 보내 파일로 받아오는 기능이기 때문에 getFile를 실행하였습니다. 

 

getFile 메서드에는 send()함수가 있는데

 

send() 함수 주석을 보면 //최초의 4바이트는 데이터 크기 라고 적혀 있는데 이 부분이 실제 구현할 때 중요합니다.

 

서버쪽에서는 사용자가 어느정도의 데이터를 보내는지 모르기때문에 잘못 구현하면 무한정 대기해야하는 상황이 

 

올 수 있습니다. (blocking으로 메서드의 제어권을 가진 함수가 데이터를 받을 때까지 메서드 호출함수는 무한정 대기)

 

최초에 데이터 크기를 알려주면 서버가 클라이언트의 요청 데이터를 무한정 대기하는 일을 없앨 수 있습니다.

 

getFile 메서드를 보시면 send후에 사용자가 보낸 파일을 read, write하여 새로운 파일로 만드는 것을 볼 수 있습니다.

 

 

 

 

 

- 파이썬 소켓 서버 소스

import socket
import os
import myCustom
import threading


server_addr = '127.0.0.1', 5555
th=[];

sema = threading.Semaphore(3)

# Create a socket with port and host bindings
def setupServer():
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    print("start sock server")
    try:
        s.bind(server_addr)
    except socket.error as msg:
        print(msg)
    return s


# Establish connection with a client
def setupConnection(s):
    s.listen(1) 
    conn, addr = s.accept()
    return conn


# Get input from user
def GET():
    reply = input("Reply: ")
    return reply


def sendFile(filename, conn):
    f = open(filename, 'rb')
    line = f.read(1024)
    while line:
        conn.send(line)
        line = f.read(1024)
    f.close()


# Loop that sends & receives data
def dataTransfer(conn, s, mode):
    while True:
    # Send a File over the network
        try:
            data = conn.recv(4);
            # 최초 4바이트는 전송할 데이터의 크기
            length = int.from_bytes(data, "little")
            #데이터 분할하여 받기
            tmpByteData=b''
            while True:
                 tmpData = conn.recv(1024)
                 tmpByteData += tmpData
                 if len(tmpByteData) == length :
                     break
            jsonData = tmpByteData.decode('utf-8')
            sema.acquire()
            filePath = myCustom.makeFile(jsonData)
            sema.release()
            sendFile(filePath, conn)
            os.remove(filePath)
            break
        except:
            break
    conn.close()


sock = setupServer()
while True:
    try:
        connection = setupConnection(sock)
    except:
        break
    client = threading.Thread(target=dataTransfer, args=(connection, sock, "SEND"))
    client.start()
    th.append(client);
    for t in th[:]:
        if not t.is_alive():
            th.remove(t)

 

 

server_addr을 보면 '127.0.0.1', 5555이라고 적혀 있는데 실서버에서 배포하실 땐

 

127.0.0.1이 아닌 private ip를 적어줘야 정상 작동하실 것입니다.

 

setupServer 함수를 보면 소켓을 생성하고 나의 ip와 실행포트번호를 bind(결합)하는 것을 볼 수 있고

 

setupConnect 함수를 보면 클라이언트의 요청을 수신(listen)하고 승인(accept)하는 것을 볼 수 있습니다.

 

맨 마지막 소스를 보면 무한 루프 while문을 볼 수 있습니다.

 

setupConnect 함수 안의 listen 함수에서 blocking이 되는 것 같습니다. 

 

클라이언트의 요청이 들어오면 함수의 제어권을 우리가 구현한 함수로 넘겨줘서 while문이 실행됩니다.

 

즉 클라이언트의 요청이 들어오지 않으면 루프문이 계속 돌고 있는게 아니라 listen에서 멈춰 있는 것입니다.

 

listen이 클라이언트의 요청을 받아야 메서드들이 다시 실행이 됩니다.

 

listen안에 있는 인자에 대해 몇몇 포스팅에서 동시 접속자 수라고 표현을 하는 것 같은데 

 

api document의 해석을 보거나 실제 실행을 해보면 동시접속자 수는 아닌 것 같습니다.

 

listen(1)을 주고 요청을 8개까지 날려보았는데 8개의 쓰레드가 생성되는 걸 보면 동시접속자 수는 아닌 것 같습니다.

 

 

api document 설명

socket.listen([backlog])

Enable a server to accept connections. If backlog is specified, it must be at least 0 (if it is lower, it is set to 0); it specifies the number of unaccepted connections that the system will allow before refusing new connections. If not specified, a default reasonable value is chosen.

 

listen 안에 있는 인자는 대기하는 요청을 몇 명까지 만들 것인지 정하는 것 같습니다.

 

예를 들어 listen(5)로 선언 하면 소켓 서버의 동시 가능한 접속자 수가 모두 꽉찼을 때

 

대기했다 다시 연결 될 수 있는 사용자의 수는 5명이고 6번째 이후로 요청한 사용자들은

 

연결을 끊어버리겠다는 의미로 해석할 수 있을 것 같습니다. 

 

즉 대기번호는 5명까지만 받고 6번째부터는 대기번호를 받지 않겠다는 얘기로 해석할 수 있을 것 같습니다.

 

무한정 대기하다 사용자 입장에서는 기다리는 시간이 오래걸릴 수 도 있으니 그런 부분을 고려할 때

 

처리할 수 있을 것 같습니다. 

 

저부분을 통해 사용자에게 '대기인원이 많습니다. 잠시 후 다시 요청해주세요.'라는 메세지를 전달 해줄 수 있는지는

 

구현을 해보지 않아서 잘 모르겠습니다.(혹시 아시는 분은 댓글로 공유해주시면 감사하겠습니다.)

 

 

listen 함수에 대해서 설명이 굉장히 길었는데 소켓에서 제공해주는 메서드이고 서버소켓에만 존재하는 메서드이니

 

중요도가 높은 것 같습니다. 

 

저도 공부하면서 listen함수에 대해서 가장 나중에 이것저것 알게된 것 같네요.

 

그 다음 dataTransfer 메서드 구현 부분을 보면 recv(4) 부분이 있습니다. 

 

이전에 자바 클라이언트에서 send할 때, 4바이트 만큼의 크기를 전달하는 데이터의 크기로 설정했습니다.

 

recv 함수도 소켓의 메서드로 함수의 제어권을 recv가 갖기 때문에 우리가 받으려는 데이터의 크기만큼 

 

recv를 해야하기 때문에 1024바이트 만큼 끊어서 루프문을 통해 모든 데이터를 다 받을 수 있도록 구현했습니다.

 

한번에 recv로 받을 수 있는 데이터의 크기에 한계가 있기 때문에 루프를 통해 끊어서 받으셔야합니다.

 

이부분 때문에 애를 좀 많이 먹었습니다. 

 

최초에 데이터의 크기를 보내줘야하는 것도 이런 부분이 있고요.

 

그 이외에는 소스를 통해 이해할 수 있는 부분일 것 같습니다.

 

동시접속자들을 처리하기 위해서 멀티쓰레드 환경으로 구현을 했는데 

 

소스를 보시면 sema = threading.Semaphore(3) 이라고 적혀있는 부분이 있습니다.

 

쓰레드를 최대 3개까지 허용하겠다는 것인데 쓰레드가 3개가 생성된다는게 아니라

 

sema.acquire()
filePath = myCustom.makeFile(jsonData)
sema.release()

 

sema.acquire()과 sema.release() 사이에 있는 소스가 실행되는 쓰레드는 3개까지라는 뜻입니다.

 

10개의 요청이 와서 쓰레드가 10개 생성되도 저 부분을 실행할 수 있는 쓰레드의 갯수는 최대 3개입니다.

 

따라서 앞선 쓰레드 3개가 먼저 할당 받고 작업이 끝나야 나머지 쓰레드들이 작업을 할당받아 실행 될 수 있습니다.

 

메모리와 cpu에 부담이 될 수 있는 부분은 3개의 쓰레드만 실행되고

 

나머지는 기다렸다 작업을 할 수 있도록 구현한 부분입니다.

 

 

이렇게 해서 자바클라이언트와 파이썬 소켓 서버의 통신을 알아봤습니다. 

 

서버 구축시 애로사항이었던 앞서 말씀 드렸던 소켓 서버 ip에 private ip로 적어야 하는 부분과

 

방화벽 부분인데

 

windows 서버의 경우 ec2의 보안그룹 인바운드 규칙 뿐만 아니라 windows 운영체제 안에서도 방화벽을 열어줘야합니다.

 

[windows defender 방화벽] -> [고급 설정]   또는 시작 검색에서 [고급 보안이 포함된 windows defender 방화벽]  클릭

 

영어 버전에서는 [windows defender firewall] -> [Advanced settings] 클릭 한 후

 

아래 절차 따라주시면 될 것 같습니다.

 

 

 

 

 

 

저는 파이썬 소켓 서버가 5555포트에서 실행했으니 사용자가 실행하는 소켓 서버의 포트 번호 입력하시면 됩니다.

 

 

 

 

 

 

 

 

 

 

 

이와 같이 실행하면 실서버에서도 정상적으로 잘 작동하실 것입니다. 

 

궁금하신 부분있으시면 댓글로 남겨주시고 잘못된 부분이나 부족한 부분 또한 댓글로 남겨주시며 감사하겠습니다.