當(dāng)前位置：首頁(yè) > 编程语言 > python >内容正文

python

java与python之间的混合开发

發(fā)布時(shí)間：2024/3/24 python 38 豆豆

生活随笔收集整理的這篇文章主要介紹了 java与python之间的混合开发小編覺(jué)得挺不錯(cuò)的,現(xiàn)在分享給大家,幫大家做個(gè)參考.

java與python之間的混合開(kāi)發(fā)

1. 項(xiàng)目需求

使用java進(jìn)行進(jìn)行數(shù)據(jù)庫(kù)的訪問(wèn)，并對(duì)查詢出的數(shù)據(jù)進(jìn)行的數(shù)據(jù)清理，使用python進(jìn)行無(wú)監(jiān)督分類

應(yīng)用場(chǎng)景：

. 整個(gè)項(xiàng)目是BS架構(gòu)，基于SpringBoot2.0

. 業(yè)務(wù)端主要是通過(guò)Java來(lái)實(shí)現(xiàn)后臺(tái)與數(shù)據(jù)庫(kù)的連接與交互

. 算法端主要通過(guò)python實(shí)現(xiàn)

實(shí)現(xiàn)的目標(biāo)：

Java通過(guò)多線程任務(wù)調(diào)度定時(shí)完成任務(wù)調(diào)度，完成對(duì)數(shù)據(jù)庫(kù)中的某一數(shù)據(jù)進(jìn)行聚類并生成對(duì)應(yīng)的聚類模型，聚類和模型生成由Python實(shí)現(xiàn)；

通過(guò)前端調(diào)用Resful接口來(lái)調(diào)取后端的接口，再通過(guò)java來(lái)與python交互，實(shí)現(xiàn)模型的調(diào)用來(lái)完成新數(shù)據(jù)的推理；

整個(gè)過(guò)程python不涉及查數(shù)據(jù)庫(kù)，Java查數(shù)據(jù)庫(kù)之后實(shí)現(xiàn)數(shù)據(jù)的清洗和預(yù)處理，按照約定的格式傳遞給python來(lái)進(jìn)行數(shù)據(jù)的算法實(shí)現(xiàn)

2.方案

可以通過(guò)udp或者tcp協(xié)議來(lái)傳遞數(shù)據(jù)

缺點(diǎn)：a. 不能異步通信；b.安全性不高；c.調(diào)試麻煩；d.數(shù)據(jù)容易丟失

通過(guò)網(wǎng)絡(luò)接口（Restful）的方式，調(diào)用python服務(wù)（通過(guò)http的POST請(qǐng)求中的Body將數(shù)據(jù)傳入，沒(méi)有測(cè)試過(guò)數(shù)據(jù)量）

通過(guò)消息中間鍵，來(lái)作為數(shù)據(jù)傳遞和消息傳遞的媒介，通過(guò)隊(duì)列的方式來(lái)傳遞數(shù)據(jù)，實(shí)現(xiàn)異步通信，并且通過(guò)中間鍵作為中介有管理系統(tǒng)，安全性高，有對(duì)應(yīng)的管理界面調(diào)試比較簡(jiǎn)單，能夠查看推送的數(shù)據(jù)是什么；使用隊(duì)列來(lái)作為日志信息的輸入和輸出；

最終確定的方案：選用方案二
使用消息中間鍵的好處：
數(shù)據(jù)不容易丟
能夠?qū)崿F(xiàn)異步通信

異步方式指消息發(fā)送方在發(fā)送消息時(shí)不必知道接收方的狀態(tài)，更無(wú)需等待接收方的回復(fù)，而接收方在收到消息時(shí)也不必知道發(fā)送方的目前狀態(tài)，更無(wú)需進(jìn)行同步的消息處理，它們之間的連接完全是松耦合的，通信是非阻塞的，這種異步通信方式是由消息中間件中的消息隊(duì)列及其服務(wù)機(jī)制保障的
能夠消峰（通過(guò)隊(duì)列的形式，防止一直訪問(wèn)服務(wù)）

用消息隊(duì)列來(lái)緩沖瞬時(shí)流量，把同步的直接調(diào)用轉(zhuǎn)換成異步的間接推送，中間通過(guò)一個(gè)隊(duì)列在一端承接瞬時(shí)的流量洪峰，在另一端平滑地將消息推送出去。

3.技術(shù)選型

Java端使用springboot2.0框架來(lái)搭建服務(wù)，使用MyBattisPlus操作數(shù)據(jù)庫(kù)

消息中間鍵：本文針對(duì)的數(shù)據(jù)量不算特別大，也不需要很高的吞吐量，所以選用穩(wěn)定且功能完備，spring集成的ActiveMQ

python端使用stomp消息傳輸框架來(lái)對(duì)ActiveMq的消息隊(duì)列進(jìn)行監(jiān)督

4. 代碼實(shí)現(xiàn)

SpringBoot2.0集成ActiveMq

添加pom文件

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-activemq</artifactId><version>1.5.0.RELEASE</version> </dependency>  <dependency><groupId>org.apache.activemq</groupId><artifactId>activemq-pool</artifactId><version>5.15.0</version> </dependency>

構(gòu)建一個(gè)生產(chǎn)者服務(wù)

java和ActiveMq相關(guān)的配置文件

通過(guò)yml文件配置

server:port: 777 #生產(chǎn)者，微服務(wù)端口號(hào) #ActiveMq spring:activemq:broker-url: tcp://192.168.1.151:61616 #mq的位置（61616是后臺(tái)的位置）user: adminpassword: adminjms:pub-sub-domain: false #false為Queue true為topic,默認(rèn)為false,目前正在探索怎么隊(duì)列和topic一起使用 processQueueName: activemq-queue-data-process reciveQueueName: activemq-queue-result trainQueueName: activemq-queue-data-train reciveFigQueueName: activemq-queue-result_fig reciveLogTopic: /topic/Log

config配置，注意**@Bean**注解的是對(duì)象的初始化函數(shù)，后面使用Autoware注入的時(shí)候，會(huì)自動(dòng)的調(diào)用這個(gè)初始化函數(shù)，生成對(duì)應(yīng)的對(duì)象

package com.atguigu.activemq.bootMqProduce.bootMqProduce.config;import org.apache.activemq.command.ActiveMQQueue; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.jms.annotation.EnableJms; import org.springframework.stereotype.Component;import javax.jms.Queue;//只要是boot，就把Component標(biāo)注在上面 //類似于Spring的applicationContext.xml @Component //讓SpringBoot管理起來(lái) @EnableJms //開(kāi)啟Jms配置 public class ConfigBean {//從yml中讀取對(duì)應(yīng)鍵值的配置注入@Value("${myqueue}")private String myQueue;//配置了一個(gè)ActiveMQ的隊(duì)列@Bean //配置Bean 有點(diǎn)兒類似于之前Spring的配置<bean id=" " class=" ">public Queue queue(){return new ActiveMQQueue(myQueue);}}

生產(chǎn)消息

package com.atguigu.activemq.bootMqProduce.bootMqProduce.produce;import org.springframework.beans.factory.annotation.Autowired; import org.springframework.jms.core.JmsMessagingTemplate; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component;import javax.jms.Queue; import java.util.UUID;//這個(gè)就相當(dāng)于impl@Component //只要是boot都要標(biāo)注 public class QueueProduce {@Autowired //這是一個(gè)通用的jms（java的消息標(biāo)準(zhǔn)）的類，里面定義了java消息的生產(chǎn)消費(fèi)流程標(biāo)準(zhǔn)private JmsMessagingTemplate jmsMessagingTemplate;@Autowired //之前在配置里面定義了一個(gè)Bean，那么就可以通過(guò)@Autowired自動(dòng)注入進(jìn)來(lái),是需要配置的private Queue queue;public void produceMsg(){jmsMessagingTemplate.convertAndSend(queue, "*****"+ UUID.randomUUID().toString().substring(0,6));//自動(dòng)轉(zhuǎn)換后寫入}@Scheduled(fixedDelay = 3000) //定時(shí)發(fā)送，每隔3000ms發(fā)送一次public void produceMsgScheduled(){jmsMessagingTemplate.convertAndSend(queue, "*****"+ UUID.randomUUID().toString().substring(0,6));//自動(dòng)轉(zhuǎn)換后寫入System.out.println("***Msg發(fā)送成功***");} }

主函數(shù)，啟動(dòng)服務(wù)

package com.atguigu.activemq.bootMqProduce.bootMqProduce;import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.scheduling.annotation.EnableScheduling;@SpringBootApplication @EnableScheduling //定時(shí)發(fā)送需要打開(kāi)這個(gè)開(kāi)關(guān) public class MainApp_produce {public static void main(String[] args) {SpringApplication.run(MainApp_produce.class,args);} }

構(gòu)建一個(gè)消費(fèi)者服務(wù)

yml配置和生產(chǎn)者一樣
config配置（無(wú)，只需要指定需要監(jiān)聽(tīng)得隊(duì)列即可）
注冊(cè)消費(fèi)者監(jiān)聽(tīng)消息隊(duì)列
package com.atguigu.activemq.bootMqConsumer.consumer;import org.springframework.beans.factory.annotation.Value; import org.springframework.jms.annotation.JmsListener; import org.springframework.stereotype.Component;import javax.jms.JMSException; import javax.jms.TextMessage;@Component //只要是boot，要么是service，要么是Component public class Queue_Consumer {//監(jiān)聽(tīng)注冊(cè),指定需要監(jiān)聽(tīng)得隊(duì)列@JmsListener(destination = "${myqueue}")public void recieve(TextMessage textMessage) throws JMSException{System.out.println("***消費(fèi)者收到消息***"+textMessage);} }
注意在這個(gè)過(guò)程中，python通過(guò)stomp協(xié)議在進(jìn)行隊(duì)列消息得傳輸過(guò)程中使用的二進(jìn)制流數(shù)進(jìn)行傳輸?shù)?#xff0c;所以，在java端接收python返回的數(shù)據(jù)的時(shí)候，需要將二進(jìn)制流先轉(zhuǎn)換為字符串，所以此處不能使用textmessage來(lái)定義接收的消息，而是使用BytesMessage來(lái)定義接收到的消息
package com.atguigu.activemq.bootMqConsumer.consumer;import org.springframework.beans.factory.annotation.Value; import org.springframework.jms.annotation.JmsListener; import org.springframework.stereotype.Component;import javax.jms.JMSException; import javax.jms.TextMessage;@Component //只要是boot，要么是service，要么是Component public class Queue_Consumer {//監(jiān)聽(tīng)注冊(cè),指定需要監(jiān)聽(tīng)得隊(duì)列@JmsListener(destination = "${myqueue}")public void recieve(TextMessage textMessage) throws JMSException{System.out.println("***消費(fèi)者收到消息***"+textMessage);} }

python端往隊(duì)列中發(fā)送消息和監(jiān)聽(tīng)隊(duì)列中的消息

使用stomp消息傳輸協(xié)議

在進(jìn)行消息消費(fèi)的時(shí)候，采用兩種模式

訓(xùn)練模式，只消費(fèi)對(duì)應(yīng)的隊(duì)列中最新的消息
推斷模式，將隊(duì)列中的數(shù)據(jù)來(lái)一條處理一條

# -*- coding: utf-8 -*- """ Created on Thu Jul 19 09:54:08 2018@author: lihc """# -*-coding:utf-8-*- import stomp import datetime import json from FINCH import * import joblib import os import tracebackqueue_receive_train = 'activemq-queue-data-train' queue_receive_process = 'activemq-queue-data-process' topic_log = '/topic/Log' listener_name = 'SampleListener' queue_send_data = 'activemq-queue-result' queue_process_result = 'activemq-process-result' queue_send_fig = 'activemq-queue-result_fig_bak' post = 61613 url = '192.168.1.151'#消息回調(diào)函數(shù)，有消息傳入的時(shí)候調(diào)用on_message函數(shù) class SampleListener(object):def on_message(self, headers, message):data_new = json.loads(message)# # print('headers: %s' % headers)print(">>>>>>>>>>>>>>>>>>>>>>>>>>>listener1")print('message: %s' % data_new)class HaddleListener(object):def __init__(self):self.flag = Falseself.data = {}self.header = ''#將不同signal_type的最新數(shù)據(jù)存入data字典中def on_message(self, headers, message):# return messageself.flag = Truedata = json.loads(message)signal_type = ''.join(list(data.keys()))self.data[signal_type] = dataself.header = headersprint(data)class ActiveMqManager(object):def __init__(self, url, post):self.url = urlself.port = post#建立連接self.con = stomp.Connection10([(self.url, self.port)])self.con.connect(username='admin', passcode='admin', wait=True)# 推送到隊(duì)列queuedef send_to_queue(self, msg, queue_name):con = stomp.Connection10([(self.url, self.port)])if not isinstance(msg, bytes):msg_json = json.dumps(msg)# 二進(jìn)制流直接傳輸?else:msg_json = msgcon.connect(username='admin', passcode='admin', wait=True)con.send(destination=queue_name, body=msg_json, content_type='text/plain')con.disconnect()# 推送到主題def send_to_topic(self, msg, topic_name):con = stomp.Connection10([(self.url, self.port)])con.connect()con.send(topic_name, msg)con.disconnect()##從隊(duì)列接收消息def receive_from_queue(self, queue_receive, listener_name, listener):con = stomp.Connection10([(self.url, self.port)])con.set_listener(listener_name, listener)con.connect(username='admin', passcode='admin', wait=True)con.subscribe(queue_receive, headers={"listener_name": listener_name})# con.subscribe(queue_receive, headers={"listener_name": listener_name}, ack='client')##從主題接收消息def receive_from_topic(self, topic_name):con = stomp.Connection10([(self.url, self.port)])con.set_listener(listener_name, SampleListener())con.connect()con.subscribe(topic_name)while 1:self.send_to_topic('topic')time.sleep(3) # secscon.disconnect()if __name__ == '__main__':activeMqManager = ActiveMqManager(url, post)train_data_listener = HaddleListener()process_data_listener = HaddleListener()# 只要主線程不結(jié)束，這個(gè)監(jiān)聽(tīng)就一直都在，也不能 disconnecion# 注冊(cè)兩個(gè)監(jiān)聽(tīng)器，分別監(jiān)聽(tīng)queue_receive_train和queue_receive_process隊(duì)列activeMqManager.receive_from_queue(queue_receive_train, listener_name="TrainListener", listener=train_data_listener)activeMqManager.receive_from_queue(queue_receive_process, listener_name="ProcessListener", listener=process_data_listener)print("主線程阻塞")while True:##對(duì)于訓(xùn)練數(shù)據(jù)的隊(duì)列，我們需要使用的是隊(duì)列中最新的一條數(shù)據(jù)，所以通過(guò)這種機(jī)制訓(xùn)練最新的一條數(shù)據(jù)""">>>>>>>>>>>>>>>>>>通過(guò)這種控制機(jī)制，讓每次程序啟動(dòng)的時(shí)候只訓(xùn)練最新的數(shù)據(jù)<<<<<<<<<<<<<<<<<<<<<<"""try:if train_data_listener.flag == True:time.sleep(8) # secs# 訓(xùn)練（只處理最新的數(shù)據(jù)）if(train_data_listener.flag == True and train_data_listener.data != {}):train_data_listener.flag = False# data = train_data_listener.data# print('message: %s' % train_data_listener.data)print(datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"), ">>>>>>>>>>>>>>>>>>>>>>處理")for key, value in train_data_listener.data.items():# 將不同的signal_type的最新的數(shù)據(jù)保存下來(lái)signal_type = ''.join(list(value.keys()))# 處理消息dataProcessing = DataProcessing(type=signal_type, train_data_dic=value)# dataProcessing = DataProcessing(type=signal_type, train_data_dic=data, model="FINCH")result_map = dataProcessing.structSignalData()#python數(shù)據(jù)處理的結(jié)果回傳給java的業(yè)務(wù)端activeMqManager.send_to_queue(result_map, queue_send_data)print("發(fā)送成功")# 通過(guò)話題來(lái)傳輸日志activeMqManager.send_to_topic("發(fā)送成功", topic_log) # 消費(fèi)完了之后，將data的數(shù)據(jù)給清空train_data_listener.data = {}# 推斷：來(lái)一條處理一條# if ( process_data_listener.data is not None):if (process_data_listener.flag == True and process_data_listener.data != {}):process_data_listener.flag = False# data = process_data_listener.dataprint(datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"), ">>>>>>>>>>>>>>>>>>>>>>測(cè)試")for key, value in process_data_listener.data.items():# 通過(guò)訂閱話題來(lái)實(shí)時(shí)傳輸日志activeMqManager.send_to_topic(">>>>>>>>>>>>>>>>>>>>>>測(cè)試", topic_log)signal_type = ''.join(list(value.keys()))if signal_type:result_map = dataProcessing.process(value)activeMqManager.send_to_topic("識(shí)別完成", topic_log)activeMqManager.send_to_queue(result_map, queue_process_result)print("識(shí)別結(jié)果發(fā)送成功")activeMqManager.send_to_topic("識(shí)別結(jié)果發(fā)送成功", topic_log)# 消費(fèi)完了之后清空process_data_listener.data = {} except (Exception) as error:print(traceback.print_exc())print("失敗代碼："+ str(error))

4.傳輸?shù)南⒌母袷?/h3>
java生產(chǎn)者生產(chǎn)消息的格式如下：
{"LD": {"LD":["PL":300,"JD":30,"WD":30]},"SX": {"SX":["PL":345,"JD":80,"WD":65]} }
數(shù)據(jù)以JSON字符串的形式進(jìn)行傳輸

Java端接收數(shù)據(jù)并解析數(shù)據(jù)的方式

消費(fèi)者的代碼

接收的數(shù)據(jù)格式為BytesMessage
@JmsListener(destination = "activemq-queue-result")@Transactionalpublic void recieve(BytesMessage bytesMessage) throws JMSException {byte[] bys = null;//解析二進(jìn)制數(shù)據(jù)為字符串?dāng)?shù)據(jù)bys = new byte[(int) bytesMessage.getBodyLength()];bytesMessage.readBytes(bys);Gson gson = new Gson();//將數(shù)據(jù)轉(zhuǎn)換為指定格式的類，Map.class或者某個(gè)結(jié)構(gòu)體Map map = gson.fromJson(new String(bys),Map.class);System.out.println("消費(fèi)者收到消息"+map.toString());System.out.println("消費(fèi)者處理消息結(jié)果------>>" + result);}
生產(chǎn)者業(yè)務(wù)代碼

業(yè)務(wù)實(shí)現(xiàn)，通過(guò)yaml的配置，來(lái)找到需要加載的表名和字段，根據(jù)約定的數(shù)據(jù)格式，轉(zhuǎn)換為json發(fā)送給MQ消息隊(duì)列

工具：Yaml2Gson，用來(lái)將yaml文件轉(zhuǎn)換為gson格式（）、Gson
public void send2Mq(String queryTable) {Map featureMap = new HashMap();Map configMap = (Map) Yaml2Gson.yamlToMap(zcYml).get(queryTable); // System.out.println(configMap.keySet().toString());featureMap.put("config",configMap );StringBuilder conditionBuilder = new StringBuilder();String features = configMap.keySet().toString();StringBuilder columnsBuilder = new StringBuilder();//拼湊出各個(gè)列的查詢//查詢的特征項(xiàng)為空的，依據(jù)其是否為連續(xù)值為其設(shè)定初始值None or 0for(String feature: features.substring(1,features.length()-1).split(", ")){Map featsMap = (Map)configMap.get(feature);if((Integer)featsMap.get("isContinuous") == 0) columnsBuilder.append("nvl(to_char("+feature+"),'None') "+feature);else columnsBuilder.append("nvl("+feature+",0) "+feature);conditionBuilder.append(feature+" is not null or ");columnsBuilder.append(", ");}String columns = columnsBuilder.substring(0,columnsBuilder.length()-2);String condition = conditionBuilder.substring(0,conditionBuilder.length()-3);List mess = combineQueryMapper.getFeatures(columns,queryTable,condition);//featureMap.put("data",mess);Map resMap = new HashMap();resMap.put(queryTable,featureMap);//數(shù)據(jù)格式如下produceQueue.produceTrainMsg(new Gson().toJson(resMap));}

5.python解析數(shù)據(jù)的方式

如果是二進(jìn)制的數(shù)據(jù)流就直接傳輸，如果是其他形式的數(shù)據(jù)，通過(guò)json.dumps將數(shù)據(jù)轉(zhuǎn)為json字符串進(jìn)行傳輸，并在java中進(jìn)行解析

if not isinstance(msg, bytes):msg_json = json.dumps(msg)# 二進(jìn)制流直接傳輸? else:msg_json = msg con.connect(username='admin', passcode='admin', wait=True) ##通過(guò)這個(gè)函數(shù)發(fā)送數(shù)據(jù)的時(shí)候，是自動(dòng)會(huì)轉(zhuǎn)換為二進(jìn)制數(shù)據(jù)流的（轉(zhuǎn)換為二進(jìn)制數(shù)據(jù)流的json格式的字符串） con.send(destination=queue_name, body=msg_json, content_type='text/plain')

6.總結(jié)

在不同架構(gòu)的代碼之間傳輸傳遞不同結(jié)構(gòu)的結(jié)構(gòu)體的數(shù)據(jù)，使用JSON字符串能夠最大程度的保留數(shù)據(jù)的原本形式。配合使用消息中間鍵通過(guò)數(shù)據(jù)流就能夠很好的將兩種不同的語(yǔ)言進(jìn)行組合使用，發(fā)揮他們各自的優(yōu)勢(shì)，并且規(guī)定好特定的數(shù)據(jù)格式之后，還能夠很容易的將兩塊的功能進(jìn)行松耦合并且達(dá)到不同模塊復(fù)用的目的。同時(shí)在使用的過(guò)程中一定要注意處理好數(shù)據(jù)的類型和數(shù)據(jù)為空的情況（根據(jù)實(shí)際的數(shù)據(jù)處理業(yè)務(wù)來(lái)將為空的數(shù)據(jù)填寫入合適的值），無(wú)論是python還是java，因?yàn)閚ull類型一般都是使用object這種大的類來(lái)向下兼容的，傳遞過(guò)來(lái)的數(shù)字都被轉(zhuǎn)換為object類型的數(shù)據(jù)。最后第一次完全原創(chuàng)的寫一篇自己的開(kāi)發(fā)總結(jié)，希望大家能夠留言指正以及私信討論！

總結(jié)

以上是生活随笔為你收集整理的java与python之间的混合开发的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。

Python
JAVA

上一篇： git 报错
下一篇：用Python实现电子邮件接收程序（PO