Java 线程池+分布式实现代码

package com.lj.demo5;
import java.util.linkedlist;
import java.util.scanner;
/**
 * 多线程与集合综合应用示例
 * 演示线程池的初始化和任务提交过程
 * 
 * 应用场景：
 * 如qq聊天服务器，面对大量用户连接时，
 * 使用线程池管理线程资源比为每个用户创建新线程更高效
 */
public class poolinit {
	// 存储待执行任务的链表，作为任务队列
	public static linkedlist<string> linktaskname = new linkedlist<string>();
	// 线程池实例
	public static threadpool t;
	public poolinit() {
		// 初始化线程池，指定核心线程数为3
		t = new threadpool(3);
		// 从控制台接收任务并提交给线程池
		scanner scanner = new scanner(system.in);
		string taskname = "";
		while (true) {
			system.out.println("请输入线程需要处理的任务名称：");
			taskname = scanner.next();
			// 将任务添加到任务队列
			linktaskname.add(taskname);
			// 提交第一个任务给线程池执行
			t.executetask(linktaskname.get(0));
		}
	}
	public static void main(string[] args) {
		// 启动线程池
		new poolinit();
	}
}

package com.lj.demo5;
import java.util.vector;
/**
 * 线程池核心类
 * 管理工作线程，负责任务的分配与调度
 */
public class threadpool {
	// 线程池大小
	private int num;
	// 存储工作线程的集合，使用vector保证线程安全
	vector<runnable> vectors ;
	/**
	 * 构造方法，初始化线程池
	 * @param num 线程池中的线程数量
	 */
	public threadpool(int num) {
		this.num = num;
		// 初始化线程集合，容量为指定的线程数量
		this.vectors = new vector<runnable>(this.num);
		system.out.println("线程集合当前大小：" + this.vectors.size());
		system.out.println("线程集合容量：" + this.vectors.capacity());
		// 创建指定数量的工作线程并启动
		for(int i=0;i<this.vectors.capacity();i++) {
			workthread w = new workthread();
			this.vectors.add(w);
			w.start(); // 启动工作线程
		}
	}
	/**
	 * 执行任务方法
	 * 查找空闲线程并分配任务
	 * @param taskname 任务名称
	 */
	public void executetask(string taskname) {
		// 遍历线程池中的线程，寻找空闲线程
		for(int i=0;i<this.vectors.capacity();i++) {
			workthread w = (workthread) this.vectors.get(i);
			system.out.println("线程池中的线程" + w.getname() + "的状态为:" + w.isflag());
			// 如果线程处于空闲状态（flag为false）
			if(!w.isflag()) {
				system.out.println(w.getname() + ",现在是空闲状态，分配新任务");
				// 设置线程为忙碌状态
				w.setflag(true);
				// 分配任务
				w.settaskname(taskname);
				// 从任务队列移除已分配的任务
				poolinit.linktaskname.remove(0);
				break; // 找到空闲线程后退出循环
			}
		}
	}
}

package com.lj.demo5;
/**
 * 工作线程类
 * 实际执行任务的线程
 */
public class workthread extends thread {
	// 线程状态标志：false表示空闲，true表示忙碌
	private boolean flag;
	// 当前执行的任务名称
	private string taskname;
	/**
	 * 线程运行方法
	 * 循环处理任务：有任务则执行，无任务则等待
	 */
	@override
	public synchronized void run() {
		while (true) {
			if (this.isflag()) { // 有任务要执行
				system.out.println(thread.currentthread().getname() + "正在执行任务：" + this.taskname);
				try {
					// 模拟任务执行时间（30秒）
					thread.sleep(30 * 1000);
				} catch (interruptedexception e) {
					e.printstacktrace();
				}
				system.out.println(thread.currentthread().getname() + "已完成任务：" + this.taskname + ",状态变为空闲");
				// 任务完成，设置为空闲状态
				this.setflag(false);
			} else { // 无任务
				system.out.println("任务队列中等待的任务数量：" + poolinit.linktaskname.size());
				// 检查是否有未处理的任务
				if (poolinit.linktaskname.size() > 0) {
					string taskname = poolinit.linktaskname.get(0);
					poolinit.t.executetask(taskname);
					system.out.println(thread.currentthread().getname() + 
							"发现未处理任务，尝试处理：" + taskname);
				} else {
					system.out.println(thread.currentthread().getname() + "：当前没有任务，进入等待状态...");
					try {
						// 进入等待状态，释放锁
						this.wait();
					} catch (interruptedexception e) {
						e.printstacktrace();
					}
				}
			}
		}
	}
	// getter和setter方法
	public boolean isflag() {
		return flag;
	}
	/**
	 * 设置线程状态
	 * 当设置为忙碌状态时，唤醒等待的线程
	 */
	public synchronized void setflag(boolean flag) {
		this.flag = flag;
		if (this.flag) {
			// 有新任务，唤醒线程
			this.notify();
		}
	}
	public string gettaskname() {
		return taskname;
	}
	public void settaskname(string taskname) {
		this.taskname = taskname;
	}
}

package com.lj.data;
import javax.jws.webmethod;
import javax.jws.webservice;
/**
 * webservice服务接口
 * 定义远程可调用的方法
 * 
 * rpc（远程过程调用）机制：
 * - rmi：客户端和服务器端都必须是java
 * - webservice：支持跨语言、跨平台通信
 * - 其他：hessian, thrift, grpc, dubbo等
 */
@webservice(targetnamespace = "http://lj.com/wsdl")
public interface iadminservice {
	/**
	 * 定义webservice方法
	 * @return 字符串结果
	 */
	@webmethod
	public string querystr();
}

package com.lj.data.impl;
import javax.jws.webservice;
import com.lj.data.iadminservice;
/**
 * webservice服务实现类
 * 提供接口中定义的方法的具体实现
 */
@webservice(
    portname = "admin",//端口名称
    servicename = "adminserviceimpl",//服务名称
    targetnamespace = "http://lj.com/wsdl",//必须与接口的命名空间一致
    endpointinterface = "com.lj.data.iadminservice"//指定实现的接口
)
public class adminserviceimpl implements iadminservice {
	/**
	 * 实现查询方法
	 * @return 返回示例字符串
	 */
	@override
	public string querystr() {
		return "你好，世界";
	}
}

package com.lj.webserviceclient;
import javax.xml.ws.endpoint;
import com.lj.data.iadminservice;
import com.lj.data.impl.adminserviceimpl;
/**
 * webservice服务发布类
 * 将服务发布到指定地址，供客户端调用
 */
public class app {
    public static void main(string[] args) {
    	system.out.println("开始发布webservice远程服务...");
    	// 创建服务实现类实例
    	iadminservice adminservice = new adminserviceimpl();
    	// 发布服务，指定服务地址
    	endpoint.publish("http://127.0.0.1:8225/adminserviceimpl/admin", adminservice);
    	system.out.println("webservice服务发布成功！");
    }
}

package com.lj.webserviceclient;
import java.net.url;
import javax.xml.namespace.qname;
import javax.xml.ws.service;
import com.lj.data.iadminservice;
/**
 * webservice客户端
 * 调用远程发布的webservice服务
 */
public class app {
    public static void main(string[] args) {
        try {
        	// 服务的url地址
            url url = new url("http://127.0.0.1:8225/adminserviceimpl/admin");
            // 创建qname对象，指定命名空间和服务名称
            qname qname = new qname("http://lj.com/wsdl", "adminserviceimpl");
            // 创建service对象
            service service = service.create(url, qname);
            // 获取服务接口实例
            iadminservice adminservice = service.getport(iadminservice.class);
            // 调用远程方法
            string message = adminservice.querystr();
            system.out.println("调用远程webservice服务的结果为: " + message);
        } catch (exception e) {
            e.printstacktrace();
        }
    }
}