diskpagestore.java

Wicket一个开发Java Web应用程序框架。它使得开发web应用程序变得容易而轻松。 Wicket利用一个POJO data beans组件使得它可以与任何持久层技术相结合。

	// contains list of serialized pages to be saved - only non empty lists
	private final Map /* <String, List<SerializedPage>> */pagesToSaveActive = new ConcurrentHashMap();

	/**
	 * Returns the list of pages to be saved for the specified session id. If the list is not found,
	 * new list is created.
	 * 
	 * @param sessionId
	 * @return
	 */
	protected List getPagesToSaveList(String sessionId)
	{
		List list = (List)pagesToSaveAll.get(sessionId);
		if (list == null)
		{
			synchronized (pagesToSaveAll)
			{
				list = (List)pagesToSaveAll.get(sessionId);
				if (list == null)
				{
					list = new ArrayList();
					pagesToSaveAll.put(sessionId, list);
				}
			}
		}
		return list;
	}

	/**
	 * Saves all entries from the specified list.
	 * 
	 * @param sessionId
	 * @param list
	 */
	protected void flushPagesToSaveList(String sessionId, List /* <SerializedPage> */list)
	{
		if (list != null)
		{
			for (Iterator i = list.iterator(); i.hasNext();)
			{
				SerializedPage page = (SerializedPage)i.next();
				getSessionEntry(sessionId, true).savePage(page);
			}
			list.clear();
		}
	}

	/**
	 * Schedules the pages to be saved by the worker thread.
	 * 
	 * @param sessionId
	 * @param pages
	 */
	private void schedulePagesSave(String sessionId, List/* <SerializedPage> */pages)
	{
		List list = getPagesToSaveList(sessionId);
		synchronized (list)
		{
			list.addAll(pages);

			if (list.size() > 0 && pagesToSaveActive.containsKey(sessionId) == false)
			{
				pagesToSaveActive.put(sessionId, list);
			}
		}
	}

	/**
	 * Worker thread that saves the serialized pages. Saving pages in the separate thread results in
	 * smoother performance under load.
	 * 
	 * @author Matej Knopp
	 */
	private class PageSavingThread implements Runnable
	{
		private volatile boolean stop = false;

		public void run()
		{
			while (stop == false)
			{
				// wait until we have something to save
				while (pagesToSaveActive.isEmpty() && stop == false)
				{
					try
					{
						Thread.sleep(getSavingThreadSleepTime());
					}
					catch (InterruptedException ignore)
					{
					}
				}

				// iterate through lists of pages to be saved
				for (Iterator i = pagesToSaveActive.entrySet().iterator(); i.hasNext();)
				{
					Map.Entry entry = (Map.Entry)i.next();
					String sessionId = (String)entry.getKey();
					List pages = (List)entry.getValue();

					synchronized (pages)
					{
						try
						{
							flushPagesToSaveList(sessionId, pages);
						}
						catch (Exception e)
						{
							log.error(
								"Error flushing serialized pages from worker thread for session " +
									sessionId, e);
						}
						i.remove();
					}
				}
			}

			stop = false;
		}

		/**
		 * Stops the worker thread.
		 */
		public void stop()
		{
			stop = true;

			// Block the calling thread until this thread has really stopped running
			while (stop)
			{
				try
				{
					Thread.sleep(getSavingThreadSleepTime());
				}
				catch (InterruptedException ignore)
				{
				}
			}
		}
	};

	/**
	 * Initializes the worker thread.
	 */
	private void initPageSavingThread()
	{
		if (isSynchronous() == false)
		{
			pageSavingThread = new PageSavingThread();
			Thread t = new Thread(pageSavingThread, "PageSavingThread-" + appName);
			t.setDaemon(true);
			t.setPriority(Thread.MAX_PRIORITY);
			t.start();
		}
	}

	private PageSavingThread pageSavingThread = null;

	/**
	 * Returns the amount time in milliseconds for the saving thread to sleep between checking
	 * whether there are pending serialized pages to be saved.
	 * 
	 * @return
	 */
	protected int getSavingThreadSleepTime()
	{
		return 100;
	}

	/**
	 * Returns whether the {@link DiskPageStore} should work in synchronous or asynchronous mode.
	 * Asynchronous mode uses a worker thread to save pages, which results in smoother performance.
	 * 
	 * @return
	 */
	protected boolean isSynchronous()
	{
		return false;
	}

	private int lastRecentlySerializedPagesCacheSize = 50;

	/**
	 * Sets the number of last recently serialized pages kept in cache. The cache is used to aid
	 * performance on session replication.
	 * 
	 * @param lastRecentlySerializedPagesCacheSize
	 */
	public void setLastRecentlySerializedPagesCacheSize(int lastRecentlySerializedPagesCacheSize)
	{
		this.lastRecentlySerializedPagesCacheSize = lastRecentlySerializedPagesCacheSize;
	}

	/**
	 * @return
	 */
	protected int getLastRecentlySerializedPagesCacheSize()
	{
		return lastRecentlySerializedPagesCacheSize;
	}

	private final SerializedPagesCache serializedPagesCache = new SerializedPagesCache(
		getLastRecentlySerializedPagesCacheSize());

	/**
	 * Strips the actual serialized page data. This is used to store
	 * {@link SerializedPageWithSession} instance in http session to reduce the memory consumption.
	 * The data can be stripped because it's already stored on disk
	 * 
	 * @param page
	 * @return
	 */
	private SerializedPageWithSession stripSerializedPage(SerializedPageWithSession page)
	{
		List pages = new ArrayList(page.pages.size());
		for (Iterator i = page.pages.iterator(); i.hasNext();)
		{
			SerializedPage sp = (SerializedPage)i.next();
			pages.add(new SerializedPage(sp.getPageId(), sp.getPageMapName(),
				sp.getVersionNumber(), sp.getAjaxVersionNumber(), null));
		}
		return new SerializedPageWithSession(page.sessionId, page.pageId, page.pageMapName,
			page.versionNumber, page.ajaxVersionNumber, pages);
	}

	private byte[] getPageData(String sessionId, int pageId, String pageMapName, int versionNumber,
		int ajaxVersionNumber)
	{
		SessionEntry entry = getSessionEntry(sessionId, false);
		if (entry != null)
		{
			byte[] data;

			if (isSynchronous())
			{
				data = entry.loadPage(pageMapName, pageId, versionNumber, ajaxVersionNumber);
			}
			else
			{
				// we need to make sure that the there are no pending pages to
				// be saved before loading a page
				List pages = getPagesToSaveList(sessionId);
				synchronized (pages)
				{
					flushPagesToSaveList(sessionId, pages);
					data = entry.loadPage(pageMapName, pageId, versionNumber, ajaxVersionNumber);
				}
			}
			return data;
		}
		else
		{
			return null;
		}
	}

	/**
	 * Loads the data stripped by
	 * {@link #stripSerializedPage(org.apache.wicket.protocol.http.pagestore.DiskPageStore.SerializedPageWithSession)}.
	 * 
	 * @param page
	 * @return
	 */
	private SerializedPageWithSession restoreStrippedSerializedPage(SerializedPageWithSession page)
	{
		List pages = new ArrayList(page.pages.size());
		for (Iterator i = page.pages.iterator(); i.hasNext();)
		{
			SerializedPage sp = (SerializedPage)i.next();
			byte data[] = getPageData(page.sessionId, sp.getPageId(), sp.getPageMapName(),
				sp.getVersionNumber(), sp.getAjaxVersionNumber());

			pages.add(new SerializedPage(sp.getPageId(), sp.getPageMapName(),
				sp.getVersionNumber(), sp.getAjaxVersionNumber(), data));
		}

		return new SerializedPageWithSession(page.sessionId, page.pageId, page.pageMapName,
			page.versionNumber, page.ajaxVersionNumber, pages);
	}


	/**
	 * {@inheritDoc}
	 */
	public Serializable prepareForSerialization(String sessionId, Object page)
	{
		SerializedPageWithSession result = null;
		if (page instanceof Page)
		{
			result = serializedPagesCache.getPage((Page)page);
			if (result == null)
			{
				List serialized = serializePage((Page)page);
				result = serializedPagesCache.storePage(sessionId, (Page)page, serialized);
			}
		}
		else if (page instanceof SerializedPageWithSession)
		{
			SerializedPageWithSession serialized = (SerializedPageWithSession)page;
			if (serialized.page.get() == SerializedPageWithSession.NO_PAGE)
			{
				// stripped page, need to restore it first
				result = restoreStrippedSerializedPage(serialized);
			}
			else
			{
				result = serialized;
			}
		}

		if (result != null)
		{
			return result;
		}
		else
		{
			return (Serializable)page;
		}
	}

	protected boolean storeAfterSessionReplication()
	{
		return true;
	}

	/**
	 * @see org.apache.wicket.protocol.http.SecondLevelCacheSessionStore.ISerializationAwarePageStore#restoreAfterSerialization(java.io.Serializable)
	 */
	public Object restoreAfterSerialization(Serializable serializable)
	{
		if (!storeAfterSessionReplication() || serializable instanceof Page)
		{
			return serializable;
		}
		else if (serializable instanceof SerializedPageWithSession)
		{
			SerializedPageWithSession page = (SerializedPageWithSession)serializable;
			if (page.page == null || page.page.get() != SerializedPageWithSession.NO_PAGE)
			{
				storeSerializedPages(page.sessionId, page.pages);
				return stripSerializedPage(page);
			}
			else
			{
				return page;
			}
		}
		else
		{
			String type = serializable != null ? serializable.getClass().getName() : null;
			throw new IllegalArgumentException("Unknown object type " + type);
		}
	}

	public Page convertToPage(Object page)
	{
		if (page instanceof Page)
		{
			return (Page)page;
		}
		else if (page instanceof SerializedPageWithSession)
		{
			SerializedPageWithSession serialized = (SerializedPageWithSession)page;

			if (serialized.page == null ||
				serialized.page.get() != SerializedPageWithSession.NO_PAGE)
			{
				storeSerializedPages(serialized.sessionId, serialized.pages);
			}

			return getPage(serialized.sessionId, serialized.pageMapName, serialized.pageId,
				serialized.versionNumber, serialized.ajaxVersionNumber);
		}
		else
		{
			String type = page != null ? page.getClass().getName() : null;
			throw new IllegalArgumentException("Unknown object type + type");
		}
	}

	/**
	 * @see org.apache.wicket.protocol.http.SecondLevelCacheSessionStore.IPageStore#containsPage(java.lang.String,
	 *      java.lang.String, int, int)
	 */
	public boolean containsPage(String sessionId, String pageMapName, int pageId, int pageVersion)
	{
		SessionEntry entry = getSessionEntry(sessionId, false);
		if (entry != null)
		{
			byte[] data;

			if (isSynchronous())
			{
				return entry.exists(pageMapName, pageId, pageVersion);
			}
			else
			{
				// we need to make sure that the there are no pending pages to
				// be saved before loading a page
				List pages = getPagesToSaveList(sessionId);
				synchronized (pages)
				{
					flushPagesToSaveList(sessionId, pages);
					return entry.exists(pageMapName, pageId, pageVersion);
				}
			}
		}
		else
		{
			return false;
		}
	}


	private static final Logger log = LoggerFactory.getLogger(DiskPageStore.class);

}