189 8069 5689
1. 首先是jdbc.properties属性文件的编写，便于数据库移植：
目前创新互联公司已为1000+的企业提供了网站建设、域名、雅安服务器托管、网站托管运营、企业网站设计、安乡网站维护等服务，公司将坚持客户导向、应用为本的策略，正道将秉承"和谐、参与、激情"的文化，与客户和合作伙伴齐心协力一起成长，共同发展。
datasource.driverClassName=oracle.jdbc.driver.OracleDriver
datasource.url=jdbc:oracle:thin:@10.6.1.11:1521:student
datasource.username=zs
datasource.password=zs
datasource.defaultAutoCommit=true
hibernate.dialect=org.hibernate.dialect.Oracle9Dialect
#当连接池中的连接耗尽的时候c3p0一次同时获取的连接数。Default: 3
c3p0.acquireIncrement=5
#初始化时获取三个连接，取值应在minPoolSize与maxPoolSize之间。Default: 3
c3p0.initialPoolSize=10
#每60秒检查所有连接池中的空闲连接。Default: 0
c3p0.idleConnectionTestPeriod=600
#-连接池中保留的最小连接数。
c3p0.minPoolSize=5
#连接池中保留的最大连接数。Default: 15
c3p0.maxPoolSize=50
#JDBC的标准参数，用以控制数据源内加载的PreparedStatements数量。但由于预缓存的statements
#属于单个connection而不是整个连接池。所以设置这个参数需要考虑到多方面的因素。
#如果maxStatements与maxStatementsPerConnection均为0，则缓存被关闭。Default: 0
c3p0.maxStatements=100
#c3p0是异步操作的，缓慢的JDBC操作通过帮助进程完成。扩展这些操作可以有效的提升性能
#通过多线程实现多个操作同时被执行。Default: 3
c3p0.numHelperThreads=10
#最大空闲时间,60秒内未使用则连接被丢弃。若为0则永不丢弃。Default: 0
c3p0.maxIdleTime=600
#hibernate.dialect=org.hibernate.dialect.SQLServerDialect
hibernate.jdbc.batch_size=25
hibernate.jdbc.fetch_size=50
hibernate.show_sql=true
hibernate.connection.release_mode=after_transaction
2. 其次是spring配置文件的数据源配置：
bean id="propertyConfigurer" class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer"
property name="location"
value/WEB-INF/classes/conf/jdbc/jdbc.properties/value
/property
/bean
bean id="dataSource" class="com.mchange.v2.c3p0.ComboPooledDataSource" destroy-method="close" dependency-check="none"
property name="driverClass"
value${datasource.driverClassName}/value
/property
property name="jdbcUrl"
value${datasource.url}/value
/property
property name="user"
value${datasource.username}/value
/property
property name="password"
value${datasource.password}/value
/property
!-- 当连接池中连接耗尽的时候c3p0一次同时获取的连接数 --
property name="acquireIncrement"
value${c3p0.acquireIncrement}/value
/property
!--初始化时获取三个连接，取值应在minPoolSize与maxPoolSize之间。Default: 3 --
property name="initialPoolSize"
value${c3p0.initialPoolSize}/value
/property
!-- 最小连接数 --
property name="minPoolSize"
value${c3p0.minPoolSize}/value
/property
!-- 最大连接数 --
property name="maxPoolSize"
value${c3p0.minPoolSize}/value
/property
!--最大空闲时间,60秒内未使用则连接被丢弃。若为0则永不丢弃。Default: 0 --
property name="maxIdleTime"
value${c3p0.maxPoolSize}/value
/property
!--每60秒检查所有连接池中的空闲连接。Default: 0 --
property name="idleConnectionTestPeriod"
value${c3p0.idleConnectionTestPeriod}/value
/property
!--JDBC的标准参数，用以控制数据源内加载的PreparedStatements数量。但由于预缓存的statements属于单个connection而不是整个连接池。
所以设置这个参数需要考虑到多方面的因素。如果maxStatements与maxStatementsPerConnection均为0，则缓存被关闭。Default: 0--
property name="maxStatements"
value${c3p0.maxStatements}/value
/property
!--c3p0是异步操作的，缓慢的JDBC操作通过帮助进程完成。扩展这些操作可以有效的提升性能通过多线程实现多个操作同时被执行。Default: 3--
property name="numHelperThreads"
楼上holly_866指出的方向是对的， 但是拼写错误。。
ORACLE的多线程体现在DML上 在操作时， 如果见到/* +*/ （平时写备注、评论块的/**/符号中有加号， 那么则表明了使用Oracle Hint. /*+ parallel(表名,并发数)*/ (有时候写作Append parallel,或者有时候直接写Append) .
这里的并发数可以省略， 也可以自己规定。 如果是省略了， 那么它的设置是DBA完成的。 参数可以在V$parameter这个view里找到。
如果单纯从开发的角度看：
ORACLE多线程可以提高某些语句查询的速度（不是一定的，取决于你的核，和服务器, 我原本有一些材料可以图示进程数和速度的关系，可惜一时找不到， 如果需要可以再联系）。具体使用时， 做几个测试 看看速率提高多少。比如我以前做数据仓库时， 一个测试要用大概27分钟， 2进程大概是23分钟。 4进程开提高到了快22（21分50多秒）分钟。 再提高进程数其实作用就递减了。
从数据库整体来看：
多线程并不是优化了你的查询速率， 而是使用了更多数据库的资源（其他用户或者进程的资源）换来你的语句速率的提高。 联系一下你的DBA， 因为很有可能你用了多进程后，从DBA的EM上会发现你资源在某时间段内用的很高，甚至会给出警告。 找DBA给你调一下进程数，或者给你建议。 还是同样的上次数据仓库的例子。 我用8进程时， DBA和我一起在看EM， 出现了一些资源占用太多的情况， 那么他要么就要调整空间，要么就要讨论下是否值得为了短短的几分钟而降低数据库的其他运行效率。 如果非要需要， 那么看是否可以将这样的数据更新放在晚上，或者数据库比较空闲的时间段。
上面讲的都是概念， 如果你想看详细一点的运算方法，可以找书看， 我这里的资料可能需要整理，也不方便， 建议自己找一下。 不需要看的太多， 一般了解则可。
Oracle JOB实现多线程插入
Sql代码
--经测试，大数据量的插入，多线程在普通磁盘执行效率反而更慢，不如单insert语句，而在磁盘阵列硬件环境下执行效率有很大的提升。
--创建表,模拟多线程插入(TT3-TT4)
DROP TABLE TT3;
DROP TABLE TT4;
CREATE TABLE TT4 AS SELECT * FROM DBA_OBJECTS WHERE 1=0;
CREATE TABLE TT3 AS SELECT * FROM DBA_OBJECTS;
--数据分批插入参数表
DROP TABLE JOB_PARMS;
CREATE TABLE JOB_PARMS
(
JOB NUMBER PRIMARY KEY,
LO_RID INT,
HI_RID INT
);
--创建插入的存储过程
CREATE OR REPLACE PROCEDURE PROC_TEST(P_JOB IN NUMBER) IS
L_REC JOB_PARMS%ROWTYPE;
BEGIN
SELECT * INTO L_REC
FROM JOB_PARMS
WHERE JOB = P_JOB;
INSERT INTO TT4
SELECT A.OWNER,
A.OBJECT_NAME,
A.SUBOBJECT_NAME,
A.OBJECT_ID,
A.DATA_OBJECT_ID,
A.OBJECT_TYPE,
A.CREATED,
A.LAST_DDL_TIME,
A.TIMESTAMP,
A.STATUS,
A.TEMPORARY,
A.GENERATED,
A.SECONDARY
FROM (SELECT ROWNUM RN, TT3.* FROM TT3 WHERE ROWNUM = L_REC.HI_RID) A
WHERE A.RN = L_REC.LO_RID;
DELETE FROM JOB_PARMS WHERE JOB = P_JOB;
COMMIT;
END;
/
---DIY 并行调度程序块
DECLARE
L_JOB NUMBER;
C_INDEX NUMBER;--插入的数量总数
S_INDEX INT:=0;--插入的开始index
E_INDEX INT:=0;--插入的结束index
CQ_INDEX INT:=20;--循环的次数
NUM_INCREASE INT:=0;--增量累加
V_I INT:=0;--计数器
BEGIN
SELECT COUNT(*) INTO C_INDEX FROM TT3;
NUM_INCREASE:= CEIL(C_INDEX/CQ_INDEX);
WHILE CQ_INDEX V_I
LOOP
V_I:=V_I+1;
S_INDEX:=1+NUM_INCREASE*(V_I-1);
IF(V_I = 20) THEN--当等于循环次数则修改结束的index
E_INDEX:= C_INDEX;
ELSE
E_INDEX:=NUM_INCREASE*V_I;
END IF;
DBMS_JOB.SUBMIT( L_JOB, 'PROC_TEST(JOB);');
INSERT INTO JOB_PARMS(JOB, LO_RID, HI_RID)
VALUES ( L_JOB, S_INDEX, E_INDEX );
END LOOP;
END;
/
--可以参考如下：
DECLARE
CURSRO C_T IS
SELECT 年份 FROM 表名;
BEGIN
FOR C_C IN C_T LOOP
BEGIN
PROCEDURE B(C_C.年份);
END;
END LOOP;
END;
windows里所有oracle事务统一由一个oracle.exe进程管理，pmon、smon等表现为oracle.exe内的线程。
linux系统里pmon、smon都是独立的进程。
线程是进程的组成部分，进程中的资源由多个线程共享。你可以把它们想象成：
进程是老大，手上有的是钞票、棍棒；
线程是他的小弟们，身无分文；
老大提供棍棒给小弟们出去办事，办事需要钱的时候由老大分配，小弟们抢来的钞票归老大统一支配。
void* OracleProcess(GPS_DATA GpsRec) // 数据库数据处理
{
interval = 0;
struct HashItem* pHash;
pHash = inithashtable(MAX_REC2);
char sql[384] = {0};
char temp[256] = {0};
char tName[10] = {0}; // 表名字
int i,k;
int j = TotalRec RATE;
double distance;
for(i=0; i j; i++)
{
sprintf(temp,"%s%f%f%f%d",gps_last[i].tid,gps_last[i].lon,gps_last[i].lat,gps_last[i].speed,gps_last[i].udate);
InsertHash(temp, pHash, MAX_REC2); // 插入最后GPS信息到hash
memset(temp,0x00,256);
}
for(i = 0; i TotalRec; i++)
{
for(k=0; kj; k++) // 查询车机是否在册
if(strcmp(GpsRec[i].tid,tid[k]) == 0)
break;
if(k j)
{
if(GpsRec[i].udate != 0.00)
{
distance = InfoUpdate(GpsRec,i); // 最新GPS数据更新
sprintf(temp,"%s%f%f%f%d",GpsRec[i].tid,GpsRec[i].lon,GpsRec[i].lat,GpsRec[i].speed,GpsRec[i].udate);
if(GetHashTablePos(temp, pHash, MAX_REC2) == -1) // 查找hash是否存在
{
if (distance 0.0001)
{
sprintf(tName,"GPS_%d_Y",tf[k]);
InsertHash(temp, pHash, MAX_REC2); // 插入
sprintf(sql,"insert into %s (id,tm_id,lon,lat, speed, utc_time, udate,mileage,DIRECTION,DISTANCE) values (seq_gps.nextVal,'%s','%f','%f','%f','%d','%d','%f','%d','%f','%d')",
tName,GpsRec[i].tid,GpsRec[i].lon,GpsRec[i].lat,GpsRec[i].speed,GpsRec[i].utime,GpsRec[i].udate,GpsRec[i].mileage,GpsRec[i].dir,distance,interval);
printf("%s\n",sql);
oci_excu(oracle_env,(text *)sql,0); // 插入数据
memset(tName,0x00,10);
}
}
memset(sql,0x00,384);
memset(temp,0x00,256);
}
}
}
memset(GpsRec,0x00,sizeof(GpsRec));
free(pHash);
pthread_exit(NULL);
}
void TcpProcess(int tfd) // 处理TCP连接上的事务
{
struct timeval ntime;
int index = 0,times,ret;
int rlen = 0,rflag = 0;
char recvbuf[513] = {0};
bzero(recvbuf,513);
while(1)
{
ret = rlen = read(tfd,recvbuf,512);
if(rlen = 0)
break;
if((rlen%32) == 0) // 32长度为标准TCP信息
{
times = 0;
ret = 5;
while(ret--)
{
if(tflag[tfd] == tfd) // 已经存在的socket
{
LOVENIX *info = (LOVENIX *)malloc(sizeof(LOVENIX));
memset(info,0x00,sizeof(LOVENIX));
if(recvbuf[times] == 0x58 || recvbuf[times] == 0x59)
ProtocolAnalysisLovenixTcp(recvbuf[times],info);
else if(recvbuf[times] == 0x24)
ProtocolAnalysisLovenixUdp(recvbuf[times],info);
sprintf(info-tid,"%s",seq[tfd]); // 合成车辆ID
DataProcess(info); // 处理GPS数据
free(info);
gettimeofday(ntime, NULL);
cntime[tfd] = ntime.tv_sec; // 更新时间
times += 32;
}
}
}
else if(rlen 32)
{
if(!rflag)
{
if((index = RegLovenix(tfd,recvbuf)) -1)
{
sprintf(seq[tfd],"%s",tid[index]); // 将对应的socket设备ID保存
gettimeofday(ntime, NULL);
sfd[tfd] = tfd;
cntime[tfd] = ntime.tv_sec;
tflag[tfd] = tfd;
rflag = 1;
}
}
}
if(rlen 512); // 已经读完
break;
memset(recvbuf,0x00,rlen);
}
}
void *TcpServer(void *arg)
{
int port = (unsigned int) arg;
int efd,i;
struct timeval ntime;
int listener, nfds, n, listen_opt = 1, lisnum;
struct sockaddr_in my_addr, their_addr;
socklen_t len = sizeof(their_addr);
lisnum = MAXLISTEN;
for(i=0; iMAX_REC; i++)
{
sfd[i] = 0;
tflag[i] = 0;
}
if ((listener = socket(PF_INET, SOCK_STREAM, 0)) == -1) // 开启 socket 监听
{
lprintf(lfd, FATAL, "TCP Socket error!\n");
exit(1);
}
else
lprintf(lfd, INFO, "TCP socket creat susscess！\n");
setsockopt(listener, SOL_SOCKET, SO_REUSEADDR, (void *) listen_opt,(int) sizeof(listen_opt)); // 设置端口多重邦定
setnonblocking(listener);
bzero(my_addr, sizeof(my_addr));
my_addr.sin_family = PF_INET;
my_addr.sin_port = htons(port);
my_addr.sin_addr.s_addr = INADDR_ANY;
if (bind(listener, (struct sockaddr *) my_addr, sizeof(struct sockaddr)) == -1)
{
lprintf(lfd, FATAL, "TCP bind error!\n");
exit(1);
}
else
lprintf(lfd, INFO, "TCP bind susscess!\n");
if (listen(listener, lisnum) == -1)
{
lprintf(lfd, FATAL, "TCP listen error!\n");
exit(1);
}
else
lprintf(lfd, INFO, "TCP listen susscess!\n");
kdpfd = epoll_create(MAXEPOLLSIZE); // 创建 epoll句柄，把监听socket加入到epoll集合里
ev.events = EPOLLIN | EPOLLET; // 注册epoll 事件
ev.data.fd = listener;
if (epoll_ctl(kdpfd, EPOLL_CTL_ADD, listener, ev) 0)
lprintf(lfd, FATAL, "EPOLL_CTL_ADD error!\n");
while (1)
{
sem_wait(sem_tcp); // 等待 sem_TCP
sem_wait(sem_tp); // 将tp值减一
nfds = epoll_wait(kdpfd, events, MAXEPOLLSIZE, 1); // 等待有事件发生
if (nfds == -1)
lprintf(lfd, FATAL,"EPOLL_WAIT error!\n");
for (n = 0; n nfds; ++n) // 处理epoll所有事件
{
if (events[n].data.fd == listener) // 如果是连接事件
{
if ((efd = accept(listener, (struct sockaddr *) their_addr,len)) 0)
{
lprintf(lfd, FATAL, "accept error!\n");
continue;
}
else
lprintf(lfd, INFO, "Client from ：%s\tSocket ID:%d\n", inet_ntoa(their_addr.sin_addr) ,efd);
setnonblocking(efd); // 设置新连接为非阻塞模式
ev.events = EPOLLIN | EPOLLET; // 注册新连接
ev.data.fd = efd;
if (epoll_ctl(kdpfd, EPOLL_CTL_ADD, efd, ev) 0) // 将新连接加入EPOLL的监听队列
lprintf(lfd, FATAL, "EPOLL_CTL_ADD error!\n");
else
{
gettimeofday(ntime, NULL);
cntime[efd] = ntime.tv_sec;
sfd[efd] = efd;
}
}
else if (events[n].events EPOLLIN)
tpool_add_work(pool, TcpProcess, (void*)events[n].data.fd); // 读取分析TCP信息
else
{
close(events[n].data.fd);
epoll_ctl(kdpfd, EPOLL_CTL_DEL, events[n].data.fd, ev);
}
}
sem_post(sem_cm);
sem_post(sem_udp);
}
close(listener);
}
int DataProcess(LOVENIX *info) // 处理GPS数据
{
if(sflag == 0 (CacheRec != TotalRec)) // 缓存1可用且没有满
{
gps_cache[CacheRec].lat = info-lat;
gps_cache[CacheRec].mileage = info-mileage;
gps_cache[CacheRec].lon = info-lon;
gps_cache[CacheRec].speed = atod(info-speed, strlen(info-speed))*0.514444444*3.6;
gps_cache[CacheRec].udate = atoi(info-udate);
gps_cache[CacheRec].utime = atoi(info-utime);
gps_cache[CacheRec].dir = atoi(info-dir);
sprintf(gps_cache[CacheRec].tid ,"%s",info-tid);
CacheRec++;
// printf("CacheRec %d\tTotalRec %d \t sflag:%d\n",CacheRec,TotalRec,sflag);
if(CacheRec == TotalRec)
{
sflag = 1;
pthread_attr_init(attr); // 初始化属性值，均设为默认值
pthread_attr_setscope(attr, PTHREAD_SCOPE_SYSTEM);
pthread_attr_setdetachstate(attr, PTHREAD_CREATE_DETACHED); // 设置线程为分离属性
if (pthread_create(thread, attr,(void*) OracleProcess,(void*)gps_cache)) // 创建数据处理线程
lprintf(lfd, FATAL, "oracle pthread_creat error!\n");
CacheRec = 0;
}
}
else if(sflag == 1 (Cache1Rec != TotalRec)) // 缓存2可用且没有满
{
gps_cache1[Cache1Rec].mileage = info-mileage;
gps_cache1[Cache1Rec].lat = info-lat;
gps_cache1[Cache1Rec].lon = info-lon;
gps_cache1[Cache1Rec].speed = atod(info-speed, strlen(info-speed))*0.514444444*3.6;
gps_cache1[Cache1Rec].udate = atoi(info-udate);
gps_cache1[Cache1Rec].utime = atoi(info-utime);
gps_cache1[Cache1Rec].dir = atoi(info-dir);
sprintf(gps_cache1[Cache1Rec].tid ,"%s",info-tid);
Cache1Rec++;
if(Cache1Rec == TotalRec)
{
sflag = 0;
pthread_attr_init(attr); // 初始化属性值，均设为默认值
pthread_attr_setscope(attr, PTHREAD_SCOPE_SYSTEM);
pthread_attr_setdetachstate(attr, PTHREAD_CREATE_DETACHED); // 设置线程为分离属性
if (pthread_create(thread, attr,(void*) OracleProcess,(void*)gps_cache1)) // 创建数据处理线程
lprintf(lfd, FATAL, "oracle pthread_creat error!\n");
Cache1Rec = 0;
}
}
else
{
lprintf(lfd, FATAL, "No cache to use!\n");
return (0);
}
return (1);
}