快捷導(dǎo)航

詳解Python如何與?java高效的交互

更新時(shí)間：2023年06月12日 09:23:51 作者：大話性能

這篇文章主要為大家介紹了詳解Python如何與java高效的交互的方法示例詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

問題背景

項(xiàng)目過程中可能會(huì)遇到，java 從數(shù)據(jù)庫取了很多數(shù)據(jù)，但 java 本身不方便處理，所以傳遞給 python 去處理，如何傳？

解決方法

調(diào)研了一些方法，譬如可以直接在 java 中調(diào)研 python，傳入?yún)?shù)的方式，或者數(shù)據(jù)先存儲(chǔ)到 excel 中，然后 python 去 excel 中讀取，但是這些方法還是存在不便利和性能會(huì)有限制。之后，我嘗試了增加個(gè) rabbitmq 消息中間件進(jìn)行不同語言之間的通信交互。

MQ 是消費(fèi) - 生產(chǎn)者模型的一個(gè)典型的代表，一端往消息隊(duì)列中不斷寫入消息，而另一端則可以讀取或者訂閱隊(duì)列中的消息。MQ 和 JMS 類似，但不同的是 JMS 是 SUN JAVA 消息中間件服務(wù)的一個(gè)標(biāo)準(zhǔn)和 API 定義，而 MQ 則是遵循了 AMQP 協(xié)議的具體實(shí)現(xiàn)和產(chǎn)品。RabbitMQ 是一個(gè)在 AMQP 基礎(chǔ)上完成的，可復(fù)用的企業(yè)消息系統(tǒng)。

上圖是簡單抽象的描述，具體到 rabbitmq 有很多詳細(xì)的概念，這里不一一詳述。當(dāng)然，我們上面也說過，rabbitmq 是 AMQP 協(xié)議的一個(gè)開源實(shí)現(xiàn)，所以其內(nèi)部實(shí)際上也是 AMQP 中的基本概念，包括：Message(消息)、Publisher(消息生產(chǎn)者)，Exchange(交換器)，Binding(綁定)，Queue(消息隊(duì)列)，Connection(網(wǎng)絡(luò)連接)，Channel(信道)，Consumer(消息消費(fèi)者)，Virtual Host(虛擬主機(jī))，Broker(消息隊(duì)列服務(wù)器)。

實(shí)際例子

生產(chǎn)者使用 java，往隊(duì)列中寫入待處理的消息。

java 代碼（即生產(chǎn)者）

public class Producer1 {
    public final static String QUEUE_NAME="data";
    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException{
        long startTime=System.currentTimeMillis(); //獲取開始時(shí)間
        //數(shù)據(jù)準(zhǔn)備
        List<testObj> list = MockData.mockData();
        //創(chuàng)建連接工廠
        ConnectionFactory factory = new ConnectionFactory();
        //設(shè)置RabbitMQ相關(guān)信息
        factory.setHost("192.168.43.211");
        factory.setUsername("test");
        factory.setPassword("123456");
        //factory.setPort(5672);
        //創(chuàng)建一個(gè)新的連接
        Connection connection = factory.newConnection();
        //創(chuàng)建一個(gè)通道
        Channel channel = connection.createChannel();
        //聲明一個(gè)隊(duì)列        
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        List<testObj> list1 = new ArrayList<testObj>();
        list1 = new MockData().mockData();
        //發(fā)送消息到隊(duì)列中
        ObjectMapper mapper=new ObjectMapper();
        String message = mapper.writeValueAsString(list1);
        channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
        //關(guān)閉通道和連接
        channel.close();
        connection.close();
        long endTime=System.currentTimeMillis(); //獲取結(jié)束時(shí)間  
        System.out.println("程序運(yùn)行時(shí)間： "+(endTime-startTime)/1000.0+"s");
    }
}

消費(fèi)者使用 python 代碼，從消息隊(duì)列中獲取 java 生成者發(fā)送的消息進(jìn)行處理。

python 代碼（即消費(fèi)者）

if __name__ == '__main__':
    starttime = datetime.datetime.now()
    # 創(chuàng)建socket鏈接
    credentials = pika.PlainCredentials('test', '123456')
    connection = pika.BlockingConnection(pika.ConnectionParameters(
     '192.168.43.211', 5672, '/', credentials))
    # 創(chuàng)建管道
    channel = connection.channel()
    # 創(chuàng)建隊(duì)列
    queue_name = 'data'
    channel.queue_declare(queue_name)
    # 如果接受到消息就調(diào)用回調(diào)函數(shù),準(zhǔn)備接受消息
    # 聲明回調(diào)函數(shù)
    def callback(ch, method, properties, body):
        message = json.loads(body.decode())
        endtime = datetime.datetime.now()
        print("拿數(shù)據(jù)時(shí)間：{}".format(endtime - starttime))
        list = message
        for i in list:
            print(i)
    channel.basic_consume(callback, queue=queue_name, no_ack=False)
    channel.start_consuming()